Cleanup parse_cmd_line()
[fio.git] / engines / net.c
CommitLineData
ed92ac0c 1/*
da751ca9
JA
2 * net engine
3 *
4 * IO engine that reads/writes to/from sockets.
5 *
ed92ac0c
JA
6 */
7#include <stdio.h>
8#include <stdlib.h>
9#include <unistd.h>
10#include <errno.h>
11#include <assert.h>
12#include <netinet/in.h>
13#include <arpa/inet.h>
14#include <netdb.h>
5fdd124a 15#include <sys/poll.h>
ed92ac0c
JA
16
17#include "../fio.h"
ed92ac0c 18
b5af8293
JA
19struct netio_data {
20 int listenfd;
21 int send_to_net;
9cce02e8
JA
22 int use_splice;
23 int pipes[2];
b5af8293
JA
24 char host[64];
25 struct sockaddr_in addr;
26};
ed92ac0c
JA
27
28static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
29{
b5af8293 30 struct netio_data *nd = td->io_ops->data;
ed92ac0c
JA
31 struct fio_file *f = io_u->file;
32
7a6499da
JA
33 /*
34 * Make sure we don't see spurious reads to a receiver, and vice versa
35 */
b5af8293
JA
36 if ((nd->send_to_net && io_u->ddir == DDIR_READ) ||
37 (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
e1161c32 38 td_verror(td, EINVAL, "bad direction");
7a6499da 39 return 1;
ed92ac0c 40 }
7a6499da 41
ed92ac0c
JA
42 if (io_u->ddir == DDIR_SYNC)
43 return 0;
44 if (io_u->offset == f->last_completed_pos)
45 return 0;
46
e01547d2
JA
47 /*
48 * If offset is different from last end position, it's a seek.
49 * As network io is purely sequential, we don't allow seeks.
50 */
e1161c32 51 td_verror(td, EINVAL, "cannot seek");
ed92ac0c
JA
52 return 1;
53}
54
cd963e18 55static int splice_io_u(int fdin, int fdout, unsigned int len)
ed92ac0c 56{
9cce02e8 57 int bytes = 0;
7a6499da 58
9cce02e8 59 while (len) {
cd963e18 60 int ret = splice(fdin, NULL, fdout, NULL, len, 0);
9cce02e8
JA
61
62 if (ret < 0) {
63 if (!bytes)
64 bytes = ret;
65
66 break;
67 } else if (!ret)
68 break;
69
70 bytes += ret;
f657a2fb 71 len -= ret;
9cce02e8
JA
72 }
73
74 return bytes;
75}
76
77/*
cd963e18 78 * Receive bytes from a socket and fill them into the internal pipe
9cce02e8 79 */
cd963e18 80static int splice_in(struct thread_data *td, struct io_u *io_u)
9cce02e8
JA
81{
82 struct netio_data *nd = td->io_ops->data;
9cce02e8 83
cd963e18 84 return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen);
9cce02e8
JA
85}
86
87/*
cd963e18 88 * Transmit 'len' bytes from the internal pipe
9cce02e8 89 */
cd963e18
JA
90static int splice_out(struct thread_data *td, struct io_u *io_u,
91 unsigned int len)
9cce02e8
JA
92{
93 struct netio_data *nd = td->io_ops->data;
cd963e18
JA
94
95 return splice_io_u(nd->pipes[0], io_u->file->fd, len);
96}
97
98static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len)
99{
9cce02e8
JA
100 struct iovec iov = {
101 .iov_base = io_u->xfer_buf,
102 .iov_len = len,
103 };
104 int bytes = 0;
105
106 while (iov.iov_len) {
cd963e18 107 int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE);
9cce02e8
JA
108
109 if (ret < 0) {
110 if (!bytes)
111 bytes = ret;
112 break;
113 } else if (!ret)
114 break;
115
116 iov.iov_len -= ret;
cd963e18 117 iov.iov_base += ret;
f657a2fb 118 bytes += ret;
9cce02e8
JA
119 }
120
121 return bytes;
cd963e18 122
9cce02e8
JA
123}
124
125/*
cd963e18 126 * vmsplice() pipe to io_u buffer
9cce02e8 127 */
cd963e18
JA
128static int vmsplice_io_u_out(struct thread_data *td, struct io_u *io_u,
129 unsigned int len)
9cce02e8
JA
130{
131 struct netio_data *nd = td->io_ops->data;
9cce02e8 132
cd963e18
JA
133 return vmsplice_io_u(io_u, nd->pipes[0], len);
134}
9cce02e8 135
cd963e18
JA
136/*
137 * vmsplice() io_u to pipe
138 */
139static int vmsplice_io_u_in(struct thread_data *td, struct io_u *io_u)
140{
141 struct netio_data *nd = td->io_ops->data;
ed92ac0c 142
cd963e18 143 return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen);
9cce02e8
JA
144}
145
cd963e18
JA
146/*
147 * splice receive - transfer socket data into a pipe using splice, then map
148 * that pipe data into the io_u using vmsplice.
149 */
9cce02e8
JA
150static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
151{
152 int ret;
153
154 ret = splice_in(td, io_u);
cd963e18
JA
155 if (ret > 0)
156 return vmsplice_io_u_out(td, io_u, ret);
9cce02e8 157
cd963e18 158 return ret;
9cce02e8
JA
159}
160
cd963e18
JA
161/*
162 * splice transmit - map data from the io_u into a pipe by using vmsplice,
163 * then transfer that pipe to a socket using splice.
164 */
9cce02e8
JA
165static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
166{
167 int ret;
168
169 ret = vmsplice_io_u_in(td, io_u);
cd963e18
JA
170 if (ret > 0)
171 return splice_out(td, io_u, ret);
9cce02e8 172
cd963e18 173 return ret;
9cce02e8
JA
174}
175
176static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
177{
178 int flags = 0;
179
180 /*
181 * if we are going to write more, set MSG_MORE
182 */
183 if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen < td->o.size)
184 flags = MSG_MORE;
185
186 return send(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
187}
188
189static int fio_netio_recv(struct io_u *io_u)
190{
191 int flags = MSG_WAITALL;
192
193 return recv(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
194}
195
196static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
197{
198 struct netio_data *nd = td->io_ops->data;
199 int ret;
200
7101d9c2
JA
201 fio_ro_check(td, io_u);
202
9cce02e8
JA
203 if (io_u->ddir == DDIR_WRITE) {
204 if (nd->use_splice)
205 ret = fio_netio_splice_out(td, io_u);
206 else
207 ret = fio_netio_send(td, io_u);
d4f12dd0 208 } else if (io_u->ddir == DDIR_READ) {
9cce02e8
JA
209 if (nd->use_splice)
210 ret = fio_netio_splice_in(td, io_u);
211 else
212 ret = fio_netio_recv(io_u);
d4f12dd0 213 } else
7a6499da 214 ret = 0; /* must be a SYNC */
ed92ac0c 215
cec6b55d 216 if (ret != (int) io_u->xfer_buflen) {
22819ec2 217 if (ret >= 0) {
cec6b55d
JA
218 io_u->resid = io_u->xfer_buflen - ret;
219 io_u->error = 0;
36167d82 220 return FIO_Q_COMPLETED;
ed92ac0c
JA
221 } else
222 io_u->error = errno;
223 }
224
36167d82 225 if (io_u->error)
e1161c32 226 td_verror(td, io_u->error, "xfer");
ed92ac0c 227
36167d82 228 return FIO_Q_COMPLETED;
ed92ac0c
JA
229}
230
b5af8293 231static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
ed92ac0c 232{
b5af8293 233 struct netio_data *nd = td->io_ops->data;
ed92ac0c 234
b5af8293
JA
235 f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
236 if (f->fd < 0) {
237 td_verror(td, errno, "socket");
238 return 1;
ed92ac0c
JA
239 }
240
b5af8293
JA
241 if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
242 td_verror(td, errno, "connect");
243 return 1;
ed92ac0c
JA
244 }
245
246 return 0;
ed92ac0c
JA
247}
248
b5af8293 249static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
5fdd124a 250{
b5af8293
JA
251 struct netio_data *nd = td->io_ops->data;
252 socklen_t socklen = sizeof(nd->addr);
5fdd124a 253 struct pollfd pfd;
b5af8293 254 int ret;
5fdd124a 255
6d86144d 256 log_info("fio: waiting for connection\n");
5fdd124a
JA
257
258 /*
259 * Accept loop. poll for incoming events, accept them. Repeat until we
260 * have all connections.
261 */
b5af8293
JA
262 while (!td->terminate) {
263 pfd.fd = nd->listenfd;
5fdd124a
JA
264 pfd.events = POLLIN;
265
266 ret = poll(&pfd, 1, -1);
267 if (ret < 0) {
268 if (errno == EINTR)
269 continue;
270
e1161c32 271 td_verror(td, errno, "poll");
5fdd124a
JA
272 break;
273 } else if (!ret)
274 continue;
275
0c09442b
JA
276 /*
277 * should be impossible
278 */
279 if (!(pfd.revents & POLLIN))
280 continue;
281
b5af8293
JA
282 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
283 if (f->fd < 0) {
284 td_verror(td, errno, "accept");
285 return 1;
286 }
287 break;
288 }
5fdd124a 289
b5af8293
JA
290 return 0;
291}
292
b5af8293
JA
293static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
294{
295 if (td_read(td))
296 return fio_netio_accept(td, f);
297 else
298 return fio_netio_connect(td, f);
299}
300
301static int fio_netio_setup_connect(struct thread_data *td, const char *host,
302 unsigned short port)
303{
304 struct netio_data *nd = td->io_ops->data;
305
306 nd->addr.sin_family = AF_INET;
307 nd->addr.sin_port = htons(port);
308
309 if (inet_aton(host, &nd->addr.sin_addr) != 1) {
310 struct hostent *hent;
311
312 hent = gethostbyname(host);
313 if (!hent) {
314 td_verror(td, errno, "gethostbyname");
315 return 1;
5fdd124a 316 }
b5af8293
JA
317
318 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
5fdd124a
JA
319 }
320
321 return 0;
322}
323
b5af8293 324static int fio_netio_setup_listen(struct thread_data *td, short port)
ed92ac0c 325{
b5af8293 326 struct netio_data *nd = td->io_ops->data;
5fdd124a 327 int fd, opt;
ed92ac0c 328
6bedbfaf 329 fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
ed92ac0c 330 if (fd < 0) {
e1161c32 331 td_verror(td, errno, "socket");
ed92ac0c
JA
332 return 1;
333 }
334
335 opt = 1;
336 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
e1161c32 337 td_verror(td, errno, "setsockopt");
ed92ac0c
JA
338 return 1;
339 }
6bedbfaf
JA
340#ifdef SO_REUSEPORT
341 if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
e1161c32 342 td_verror(td, errno, "setsockopt");
6bedbfaf
JA
343 return 1;
344 }
345#endif
ed92ac0c 346
b5af8293
JA
347 nd->addr.sin_family = AF_INET;
348 nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
349 nd->addr.sin_port = htons(port);
ed92ac0c 350
b5af8293 351 if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
e1161c32 352 td_verror(td, errno, "bind");
ed92ac0c
JA
353 return 1;
354 }
355 if (listen(fd, 1) < 0) {
e1161c32 356 td_verror(td, errno, "listen");
ed92ac0c
JA
357 return 1;
358 }
359
b5af8293
JA
360 nd->listenfd = fd;
361 return 0;
ed92ac0c
JA
362}
363
9bec88e1 364static int fio_netio_init(struct thread_data *td)
ed92ac0c 365{
b5af8293 366 struct netio_data *nd = td->io_ops->data;
e01547d2 367 unsigned short port;
b5af8293 368 char host[64], buf[128];
ed92ac0c 369 char *sep;
af52b345 370 int ret;
ed92ac0c 371
413dd459 372 if (td_rw(td)) {
ed92ac0c
JA
373 log_err("fio: network connections must be read OR write\n");
374 return 1;
375 }
16d55aae
JA
376 if (td_random(td)) {
377 log_err("fio: network IO can't be random\n");
378 return 1;
379 }
ed92ac0c 380
2dc1bbeb 381 strcpy(buf, td->o.filename);
ed92ac0c 382
9f9214f2 383 sep = strchr(buf, '/');
ed92ac0c 384 if (!sep) {
2dc1bbeb 385 log_err("fio: bad network host/port <<%s>>\n", td->o.filename);
ed92ac0c
JA
386 return 1;
387 }
388
389 *sep = '\0';
390 sep++;
391 strcpy(host, buf);
e01547d2 392 port = atoi(sep);
ed92ac0c 393
413dd459 394 if (td_read(td)) {
b5af8293 395 nd->send_to_net = 0;
ed92ac0c
JA
396 ret = fio_netio_setup_listen(td, port);
397 } else {
b5af8293 398 nd->send_to_net = 1;
ed92ac0c
JA
399 ret = fio_netio_setup_connect(td, host, port);
400 }
401
7bb48f84 402 return ret;
ed92ac0c
JA
403}
404
b5af8293 405static void fio_netio_cleanup(struct thread_data *td)
9bec88e1 406{
b5af8293
JA
407 struct netio_data *nd = td->io_ops->data;
408
409 if (nd) {
64b24cd8
JA
410 if (nd->listenfd != -1)
411 close(nd->listenfd);
412 if (nd->pipes[0] != -1)
413 close(nd->pipes[0]);
414 if (nd->pipes[1] != -1)
415 close(nd->pipes[1]);
416
b5af8293
JA
417 free(nd);
418 td->io_ops->data = NULL;
419 }
420}
421
422static int fio_netio_setup(struct thread_data *td)
423{
7bb48f84 424 struct netio_data *nd;
7bb48f84
JA
425
426 if (!td->io_ops->data) {
427 nd = malloc(sizeof(*nd));;
428
429 memset(nd, 0, sizeof(*nd));
430 nd->listenfd = -1;
64b24cd8 431 nd->pipes[0] = nd->pipes[1] = -1;
7bb48f84 432 td->io_ops->data = nd;
7bb48f84 433 }
b5af8293 434
9bec88e1
JA
435 return 0;
436}
437
9cce02e8
JA
438static int fio_netio_setup_splice(struct thread_data *td)
439{
440 struct netio_data *nd;
441
442 fio_netio_setup(td);
443
444 nd = td->io_ops->data;
445 if (nd) {
446 if (pipe(nd->pipes) < 0)
447 return 1;
448
449 nd->use_splice = 1;
450 return 0;
451 }
452
453 return 1;
454}
455
456static struct ioengine_ops ioengine_rw = {
ed92ac0c
JA
457 .name = "net",
458 .version = FIO_IOOPS_VERSION,
ed92ac0c
JA
459 .prep = fio_netio_prep,
460 .queue = fio_netio_queue,
ed92ac0c 461 .setup = fio_netio_setup,
9bec88e1 462 .init = fio_netio_init,
b5af8293
JA
463 .cleanup = fio_netio_cleanup,
464 .open_file = fio_netio_open_file,
465 .close_file = generic_close_file,
dc372f07 466 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR,
ed92ac0c
JA
467};
468
9cce02e8
JA
469static struct ioengine_ops ioengine_splice = {
470 .name = "netsplice",
471 .version = FIO_IOOPS_VERSION,
472 .prep = fio_netio_prep,
473 .queue = fio_netio_queue,
474 .setup = fio_netio_setup_splice,
475 .init = fio_netio_init,
476 .cleanup = fio_netio_cleanup,
477 .open_file = fio_netio_open_file,
478 .close_file = generic_close_file,
b67740d3 479 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR,
9cce02e8
JA
480};
481
ed92ac0c
JA
482static void fio_init fio_netio_register(void)
483{
9cce02e8
JA
484 register_ioengine(&ioengine_rw);
485 register_ioengine(&ioengine_splice);
ed92ac0c
JA
486}
487
488static void fio_exit fio_netio_unregister(void)
489{
9cce02e8
JA
490 unregister_ioengine(&ioengine_rw);
491 unregister_ioengine(&ioengine_splice);
ed92ac0c 492}