[PATCH] syslet engine: use pread instead lseek+read
[fio.git] / engines / net.c
CommitLineData
ed92ac0c 1/*
d4f12dd0 2 * Transfer data over the net.
ed92ac0c
JA
3 */
4#include <stdio.h>
5#include <stdlib.h>
6#include <unistd.h>
7#include <errno.h>
8#include <assert.h>
9#include <netinet/in.h>
10#include <arpa/inet.h>
11#include <netdb.h>
5fdd124a 12#include <sys/poll.h>
ed92ac0c
JA
13
14#include "../fio.h"
15#include "../os.h"
16
17struct net_data {
18 int send_to_net;
19 struct io_u *last_io_u;
20};
21
22static int fio_netio_getevents(struct thread_data *td, int fio_unused min,
23 int max, struct timespec fio_unused *t)
24{
25 assert(max <= 1);
26
27 /*
28 * we can only have one finished io_u for sync io, since the depth
29 * is always 1
30 */
31 if (list_empty(&td->io_u_busylist))
32 return 0;
33
34 return 1;
35}
36
37static struct io_u *fio_netio_event(struct thread_data *td, int event)
38{
39 struct net_data *nd = td->io_ops->data;
40
41 assert(event == 0);
42
43 return nd->last_io_u;
44}
45
46static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
47{
48 struct net_data *nd = td->io_ops->data;
49 struct fio_file *f = io_u->file;
50
7a6499da
JA
51 /*
52 * Make sure we don't see spurious reads to a receiver, and vice versa
53 */
54 if ((nd->send_to_net && io_u->ddir == DDIR_READ) ||
55 (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
56 printf("boo!\n");
57 td_verror(td, EINVAL);
58 return 1;
ed92ac0c 59 }
7a6499da 60
ed92ac0c
JA
61 if (io_u->ddir == DDIR_SYNC)
62 return 0;
63 if (io_u->offset == f->last_completed_pos)
64 return 0;
65
e01547d2
JA
66 /*
67 * If offset is different from last end position, it's a seek.
68 * As network io is purely sequential, we don't allow seeks.
69 */
ed92ac0c
JA
70 td_verror(td, EINVAL);
71 return 1;
72}
73
74static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
75{
76 struct net_data *nd = td->io_ops->data;
77 struct fio_file *f = io_u->file;
d4f12dd0 78 int ret, flags = 0;
7a6499da
JA
79
80 if (io_u->ddir == DDIR_WRITE) {
7a6499da
JA
81 /*
82 * if we are going to write more, set MSG_MORE
83 */
84 if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
85 td->io_size)
86 flags = MSG_MORE;
ed92ac0c 87
7a6499da 88 ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
d4f12dd0
JA
89 } else if (io_u->ddir == DDIR_READ) {
90 flags = MSG_WAITALL;
91 ret = recv(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
92 } else
7a6499da 93 ret = 0; /* must be a SYNC */
ed92ac0c 94
cec6b55d 95 if (ret != (int) io_u->xfer_buflen) {
ed92ac0c 96 if (ret > 0) {
cec6b55d
JA
97 io_u->resid = io_u->xfer_buflen - ret;
98 io_u->error = 0;
99 return ret;
ed92ac0c
JA
100 } else
101 io_u->error = errno;
102 }
103
104 if (!io_u->error)
105 nd->last_io_u = io_u;
95bcd815
JA
106 else
107 td_verror(td, io_u->error);
ed92ac0c
JA
108
109 return io_u->error;
110}
111
112static int fio_netio_setup_connect(struct thread_data *td, const char *host,
e01547d2 113 unsigned short port)
ed92ac0c
JA
114{
115 struct sockaddr_in addr;
116 struct fio_file *f;
2fc26981 117 int i;
ed92ac0c
JA
118
119 memset(&addr, 0, sizeof(addr));
120 addr.sin_family = AF_INET;
e01547d2 121 addr.sin_port = htons(port);
ed92ac0c
JA
122
123 if (inet_aton(host, &addr.sin_addr) != 1) {
7a6499da 124 struct hostent *hent;
ed92ac0c 125
7a6499da 126 hent = gethostbyname(host);
ed92ac0c 127 if (!hent) {
6bedbfaf 128 td_verror(td, errno);
ed92ac0c
JA
129 return 1;
130 }
131
132 memcpy(&addr.sin_addr, hent->h_addr, 4);
133 }
134
2fc26981 135 for_each_file(td, f, i) {
6bedbfaf 136 f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
2fc26981 137 if (f->fd < 0) {
6bedbfaf 138 td_verror(td, errno);
2fc26981
JA
139 return 1;
140 }
ed92ac0c 141
2fc26981 142 if (connect(f->fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
6bedbfaf 143 td_verror(td, errno);
2fc26981
JA
144 return 1;
145 }
ed92ac0c
JA
146 }
147
148 return 0;
149
150}
151
5fdd124a
JA
152static int fio_netio_accept_connections(struct thread_data *td, int fd,
153 struct sockaddr_in *addr)
154{
155 socklen_t socklen = sizeof(*addr);
156 unsigned int accepts = 0;
157 struct pollfd pfd;
158
159 fprintf(f_out, "fio: waiting for %u connections\n", td->nr_files);
160
161 /*
162 * Accept loop. poll for incoming events, accept them. Repeat until we
163 * have all connections.
164 */
165 while (!td->terminate && accepts < td->nr_files) {
166 struct fio_file *f;
167 int ret, i;
168
169 pfd.fd = fd;
170 pfd.events = POLLIN;
171
172 ret = poll(&pfd, 1, -1);
173 if (ret < 0) {
174 if (errno == EINTR)
175 continue;
176
177 td_verror(td, errno);
178 break;
179 } else if (!ret)
180 continue;
181
0c09442b
JA
182 /*
183 * should be impossible
184 */
185 if (!(pfd.revents & POLLIN))
186 continue;
187
5fdd124a
JA
188 for_each_file(td, f, i) {
189 if (f->fd != -1)
190 continue;
191
192 f->fd = accept(fd, (struct sockaddr *) addr, &socklen);
193 if (f->fd < 0) {
194 td_verror(td, errno);
195 return 1;
196 }
197 accepts++;
198 break;
199 }
200 }
201
202 return 0;
203}
204
e01547d2 205static int fio_netio_setup_listen(struct thread_data *td, unsigned short port)
ed92ac0c
JA
206{
207 struct sockaddr_in addr;
5fdd124a 208 int fd, opt;
ed92ac0c 209
6bedbfaf 210 fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
ed92ac0c 211 if (fd < 0) {
6bedbfaf 212 td_verror(td, errno);
ed92ac0c
JA
213 return 1;
214 }
215
216 opt = 1;
217 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
6bedbfaf 218 td_verror(td, errno);
ed92ac0c
JA
219 return 1;
220 }
6bedbfaf
JA
221#ifdef SO_REUSEPORT
222 if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
223 td_verror(td, errno);
224 return 1;
225 }
226#endif
ed92ac0c
JA
227
228 memset(&addr, 0, sizeof(addr));
229 addr.sin_family = AF_INET;
230 addr.sin_addr.s_addr = htonl(INADDR_ANY);
e01547d2 231 addr.sin_port = htons(port);
ed92ac0c
JA
232
233 if (bind(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
6bedbfaf 234 td_verror(td, errno);
ed92ac0c
JA
235 return 1;
236 }
237 if (listen(fd, 1) < 0) {
6bedbfaf 238 td_verror(td, errno);
ed92ac0c
JA
239 return 1;
240 }
241
5fdd124a 242 return fio_netio_accept_connections(td, fd, &addr);
ed92ac0c
JA
243}
244
245static int fio_netio_setup(struct thread_data *td)
246{
e01547d2 247 char host[64], buf[128];
ed92ac0c 248 struct net_data *nd;
e01547d2 249 unsigned short port;
2fc26981 250 struct fio_file *f;
ed92ac0c 251 char *sep;
2fc26981 252 int ret, i;
ed92ac0c 253
7a6499da
JA
254 if (!td->total_file_size) {
255 log_err("fio: need size= set\n");
256 return 1;
257 }
258
ed92ac0c
JA
259 /*
260 * work around for late init call
261 */
262 if (td->io_ops->init(td))
263 return 1;
264
265 nd = td->io_ops->data;
266
267 if (td->iomix) {
268 log_err("fio: network connections must be read OR write\n");
269 return 1;
270 }
ed92ac0c
JA
271
272 strcpy(buf, td->filename);
273
274 sep = strchr(buf, ':');
275 if (!sep) {
276 log_err("fio: bad network host:port <<%s>>\n", td->filename);
277 return 1;
278 }
279
280 *sep = '\0';
281 sep++;
282 strcpy(host, buf);
e01547d2 283 port = atoi(sep);
ed92ac0c 284
85eb1d44 285 if (td->ddir == DDIR_READ) {
ed92ac0c
JA
286 nd->send_to_net = 0;
287 ret = fio_netio_setup_listen(td, port);
288 } else {
289 nd->send_to_net = 1;
290 ret = fio_netio_setup_connect(td, host, port);
291 }
292
2fc26981
JA
293 if (ret)
294 return ret;
295
296 td->io_size = td->total_file_size;
297 td->total_io_size = td->io_size;
298
299 for_each_file(td, f, i) {
300 f->file_size = td->total_file_size / td->nr_files;
301 f->real_file_size = f->file_size;
ed92ac0c
JA
302 }
303
2fc26981 304 return 0;
ed92ac0c
JA
305}
306
307static void fio_netio_cleanup(struct thread_data *td)
308{
309 if (td->io_ops->data) {
310 free(td->io_ops->data);
311 td->io_ops->data = NULL;
312 }
313}
314
315static int fio_netio_init(struct thread_data *td)
316{
317 struct net_data *nd;
318
e01547d2
JA
319 /*
320 * Hack to work-around the ->setup() function calling init on its
321 * own, since it needs ->io_ops->data to be set up.
322 */
ed92ac0c
JA
323 if (td->io_ops->data)
324 return 0;
325
326 nd = malloc(sizeof(*nd));
327 nd->last_io_u = NULL;
328 td->io_ops->data = nd;
329 return 0;
330}
331
332static struct ioengine_ops ioengine = {
333 .name = "net",
334 .version = FIO_IOOPS_VERSION,
335 .init = fio_netio_init,
336 .prep = fio_netio_prep,
337 .queue = fio_netio_queue,
338 .getevents = fio_netio_getevents,
339 .event = fio_netio_event,
340 .cleanup = fio_netio_cleanup,
341 .setup = fio_netio_setup,
342 .flags = FIO_SYNCIO | FIO_NETIO,
343};
344
345static void fio_init fio_netio_register(void)
346{
347 register_ioengine(&ioengine);
348}
349
350static void fio_exit fio_netio_unregister(void)
351{
352 unregister_ioengine(&ioengine);
353}