[PATCH] syslet engine: use pread instead lseek+read
[fio.git] / engines / net.c
1 /*
2  * Transfer data over the net.
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>
12 #include <sys/poll.h>
13
14 #include "../fio.h"
15 #include "../os.h"
16
17 struct net_data {
18         int send_to_net;
19         struct io_u *last_io_u;
20 };
21
22 static 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
37 static 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
46 static 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
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;
59         }
60                 
61         if (io_u->ddir == DDIR_SYNC)
62                 return 0;
63         if (io_u->offset == f->last_completed_pos)
64                 return 0;
65
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          */
70         td_verror(td, EINVAL);
71         return 1;
72 }
73
74 static 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;
78         int ret, flags = 0;
79
80         if (io_u->ddir == DDIR_WRITE) {
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;
87
88                 ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
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
93                 ret = 0;        /* must be a SYNC */
94
95         if (ret != (int) io_u->xfer_buflen) {
96                 if (ret > 0) {
97                         io_u->resid = io_u->xfer_buflen - ret;
98                         io_u->error = 0;
99                         return ret;
100                 } else
101                         io_u->error = errno;
102         }
103
104         if (!io_u->error)
105                 nd->last_io_u = io_u;
106         else
107                 td_verror(td, io_u->error);
108
109         return io_u->error;
110 }
111
112 static int fio_netio_setup_connect(struct thread_data *td, const char *host,
113                                    unsigned short port)
114 {
115         struct sockaddr_in addr;
116         struct fio_file *f;
117         int i;
118
119         memset(&addr, 0, sizeof(addr));
120         addr.sin_family = AF_INET;
121         addr.sin_port = htons(port);
122
123         if (inet_aton(host, &addr.sin_addr) != 1) {
124                 struct hostent *hent;
125
126                 hent = gethostbyname(host);
127                 if (!hent) {
128                         td_verror(td, errno);
129                         return 1;
130                 }
131
132                 memcpy(&addr.sin_addr, hent->h_addr, 4);
133         }
134
135         for_each_file(td, f, i) {
136                 f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
137                 if (f->fd < 0) {
138                         td_verror(td, errno);
139                         return 1;
140                 }
141
142                 if (connect(f->fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
143                         td_verror(td, errno);
144                         return 1;
145                 }
146         }
147
148         return 0;
149
150 }
151
152 static 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
182                 /*
183                  * should be impossible
184                  */
185                 if (!(pfd.revents & POLLIN))
186                         continue;
187
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
205 static int fio_netio_setup_listen(struct thread_data *td, unsigned short port)
206 {
207         struct sockaddr_in addr;
208         int fd, opt;
209
210         fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
211         if (fd < 0) {
212                 td_verror(td, errno);
213                 return 1;
214         }
215
216         opt = 1;
217         if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
218                 td_verror(td, errno);
219                 return 1;
220         }
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
227
228         memset(&addr, 0, sizeof(addr));
229         addr.sin_family = AF_INET;
230         addr.sin_addr.s_addr = htonl(INADDR_ANY);
231         addr.sin_port = htons(port);
232
233         if (bind(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
234                 td_verror(td, errno);
235                 return 1;
236         }
237         if (listen(fd, 1) < 0) {
238                 td_verror(td, errno);
239                 return 1;
240         }
241
242         return fio_netio_accept_connections(td, fd, &addr);
243 }
244
245 static int fio_netio_setup(struct thread_data *td)
246 {
247         char host[64], buf[128];
248         struct net_data *nd;
249         unsigned short port;
250         struct fio_file *f;
251         char *sep;
252         int ret, i;
253
254         if (!td->total_file_size) {
255                 log_err("fio: need size= set\n");
256                 return 1;
257         }
258
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         }
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);
283         port = atoi(sep);
284
285         if (td->ddir == DDIR_READ) {
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
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;
302         }
303
304         return 0;
305 }
306
307 static 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
315 static int fio_netio_init(struct thread_data *td)
316 {
317         struct net_data *nd;
318
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          */
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
332 static 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
345 static void fio_init fio_netio_register(void)
346 {
347         register_ioengine(&ioengine);
348 }
349
350 static void fio_exit fio_netio_unregister(void)
351 {
352         unregister_ioengine(&ioengine);
353 }