4 * IO engine that reads/writes to/from sockets.
12 #include <netinet/in.h>
13 #include <arpa/inet.h>
24 struct sockaddr_in addr;
27 static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
29 struct netio_data *nd = td->io_ops->data;
30 struct fio_file *f = io_u->file;
33 * Make sure we don't see spurious reads to a receiver, and vice versa
35 if ((nd->send_to_net && io_u->ddir == DDIR_READ) ||
36 (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
37 td_verror(td, EINVAL, "bad direction");
41 if (io_u->ddir == DDIR_SYNC)
43 if (io_u->offset == f->last_completed_pos)
47 * If offset is different from last end position, it's a seek.
48 * As network io is purely sequential, we don't allow seeks.
50 td_verror(td, EINVAL, "cannot seek");
54 static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
56 struct fio_file *f = io_u->file;
59 if (io_u->ddir == DDIR_WRITE) {
61 * if we are going to write more, set MSG_MORE
63 if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
67 ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
68 } else if (io_u->ddir == DDIR_READ) {
70 ret = recv(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
72 ret = 0; /* must be a SYNC */
74 if (ret != (int) io_u->xfer_buflen) {
76 io_u->resid = io_u->xfer_buflen - ret;
78 return FIO_Q_COMPLETED;
84 td_verror(td, io_u->error, "xfer");
86 return FIO_Q_COMPLETED;
89 static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
91 struct netio_data *nd = td->io_ops->data;
93 f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
95 td_verror(td, errno, "socket");
99 if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
100 td_verror(td, errno, "connect");
107 static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
109 struct netio_data *nd = td->io_ops->data;
110 socklen_t socklen = sizeof(nd->addr);
114 fprintf(f_out, "fio: waiting for connection\n");
117 * Accept loop. poll for incoming events, accept them. Repeat until we
118 * have all connections.
120 while (!td->terminate) {
121 pfd.fd = nd->listenfd;
124 ret = poll(&pfd, 1, -1);
129 td_verror(td, errno, "poll");
135 * should be impossible
137 if (!(pfd.revents & POLLIN))
140 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
142 td_verror(td, errno, "accept");
152 static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
155 return fio_netio_accept(td, f);
157 return fio_netio_connect(td, f);
160 static int fio_netio_setup_connect(struct thread_data *td, const char *host,
163 struct netio_data *nd = td->io_ops->data;
165 nd->addr.sin_family = AF_INET;
166 nd->addr.sin_port = htons(port);
168 if (inet_aton(host, &nd->addr.sin_addr) != 1) {
169 struct hostent *hent;
171 hent = gethostbyname(host);
173 td_verror(td, errno, "gethostbyname");
177 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
183 static int fio_netio_setup_listen(struct thread_data *td, short port)
185 struct netio_data *nd = td->io_ops->data;
188 fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
190 td_verror(td, errno, "socket");
195 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
196 td_verror(td, errno, "setsockopt");
200 if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
201 td_verror(td, errno, "setsockopt");
206 nd->addr.sin_family = AF_INET;
207 nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
208 nd->addr.sin_port = htons(port);
210 if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
211 td_verror(td, errno, "bind");
214 if (listen(fd, 1) < 0) {
215 td_verror(td, errno, "listen");
223 static int fio_netio_init(struct thread_data *td)
225 struct netio_data *nd = td->io_ops->data;
228 char host[64], buf[128];
233 if (!td->total_file_size) {
234 log_err("fio: need size= set\n");
239 log_err("fio: network connections must be read OR write\n");
243 strcpy(buf, td->filename);
245 sep = strchr(buf, '/');
247 log_err("fio: bad network host/port <<%s>>\n", td->filename);
258 ret = fio_netio_setup_listen(td, port);
261 ret = fio_netio_setup_connect(td, host, port);
267 td->io_size = td->total_file_size;
268 td->total_io_size = td->io_size;
270 for_each_file(td, f, i) {
271 f->file_size = td->total_file_size / td->nr_files;
272 f->real_file_size = f->file_size;
278 static void fio_netio_cleanup(struct thread_data *td)
280 struct netio_data *nd = td->io_ops->data;
284 td->io_ops->data = NULL;
288 static int fio_netio_setup(struct thread_data *td)
290 struct netio_data *nd = malloc(sizeof(*nd));
292 memset(nd, 0, sizeof(*nd));
294 td->io_ops->data = nd;
298 static struct ioengine_ops ioengine = {
300 .version = FIO_IOOPS_VERSION,
301 .prep = fio_netio_prep,
302 .queue = fio_netio_queue,
303 .setup = fio_netio_setup,
304 .init = fio_netio_init,
305 .cleanup = fio_netio_cleanup,
306 .open_file = fio_netio_open_file,
307 .close_file = generic_close_file,
308 .flags = FIO_SYNCIO | FIO_DISKLESSIO,
311 static void fio_init fio_netio_register(void)
313 register_ioengine(&ioengine);
316 static void fio_exit fio_netio_unregister(void)
318 unregister_ioengine(&ioengine);