dlm: Remove superfluous call to recalc_sigpending()
[linux-3.10.git] / fs / dlm / lowcomms.c
1 /******************************************************************************
2 *******************************************************************************
3 **
4 **  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
5 **  Copyright (C) 2004-2009 Red Hat, Inc.  All rights reserved.
6 **
7 **  This copyrighted material is made available to anyone wishing to use,
8 **  modify, copy, or redistribute it subject to the terms and conditions
9 **  of the GNU General Public License v.2.
10 **
11 *******************************************************************************
12 ******************************************************************************/
13
14 /*
15  * lowcomms.c
16  *
17  * This is the "low-level" comms layer.
18  *
19  * It is responsible for sending/receiving messages
20  * from other nodes in the cluster.
21  *
22  * Cluster nodes are referred to by their nodeids. nodeids are
23  * simply 32 bit numbers to the locking module - if they need to
24  * be expanded for the cluster infrastructure then that is its
25  * responsibility. It is this layer's
26  * responsibility to resolve these into IP address or
27  * whatever it needs for inter-node communication.
28  *
29  * The comms level is two kernel threads that deal mainly with
30  * the receiving of messages from other nodes and passing them
31  * up to the mid-level comms layer (which understands the
32  * message format) for execution by the locking core, and
33  * a send thread which does all the setting up of connections
34  * to remote nodes and the sending of data. Threads are not allowed
35  * to send their own data because it may cause them to wait in times
36  * of high load. Also, this way, the sending thread can collect together
37  * messages bound for one node and send them in one block.
38  *
39  * lowcomms will choose to use either TCP or SCTP as its transport layer
40  * depending on the configuration variable 'protocol'. This should be set
41  * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
42  * cluster-wide mechanism as it must be the same on all nodes of the cluster
43  * for the DLM to function.
44  *
45  */
46
47 #include <asm/ioctls.h>
48 #include <net/sock.h>
49 #include <net/tcp.h>
50 #include <linux/pagemap.h>
51 #include <linux/file.h>
52 #include <linux/mutex.h>
53 #include <linux/sctp.h>
54 #include <linux/slab.h>
55 #include <net/sctp/user.h>
56 #include <net/ipv6.h>
57
58 #include "dlm_internal.h"
59 #include "lowcomms.h"
60 #include "midcomms.h"
61 #include "config.h"
62
63 #define NEEDED_RMEM (4*1024*1024)
64 #define CONN_HASH_SIZE 32
65
66 /* Number of messages to send before rescheduling */
67 #define MAX_SEND_MSG_COUNT 25
68
69 struct cbuf {
70         unsigned int base;
71         unsigned int len;
72         unsigned int mask;
73 };
74
75 static void cbuf_add(struct cbuf *cb, int n)
76 {
77         cb->len += n;
78 }
79
80 static int cbuf_data(struct cbuf *cb)
81 {
82         return ((cb->base + cb->len) & cb->mask);
83 }
84
85 static void cbuf_init(struct cbuf *cb, int size)
86 {
87         cb->base = cb->len = 0;
88         cb->mask = size-1;
89 }
90
91 static void cbuf_eat(struct cbuf *cb, int n)
92 {
93         cb->len  -= n;
94         cb->base += n;
95         cb->base &= cb->mask;
96 }
97
98 static bool cbuf_empty(struct cbuf *cb)
99 {
100         return cb->len == 0;
101 }
102
103 struct connection {
104         struct socket *sock;    /* NULL if not connected */
105         uint32_t nodeid;        /* So we know who we are in the list */
106         struct mutex sock_mutex;
107         unsigned long flags;
108 #define CF_READ_PENDING 1
109 #define CF_WRITE_PENDING 2
110 #define CF_CONNECT_PENDING 3
111 #define CF_INIT_PENDING 4
112 #define CF_IS_OTHERCON 5
113 #define CF_CLOSE 6
114 #define CF_APP_LIMITED 7
115         struct list_head writequeue;  /* List of outgoing writequeue_entries */
116         spinlock_t writequeue_lock;
117         int (*rx_action) (struct connection *); /* What to do when active */
118         void (*connect_action) (struct connection *);   /* What to do to connect */
119         struct page *rx_page;
120         struct cbuf cb;
121         int retries;
122 #define MAX_CONNECT_RETRIES 3
123         int sctp_assoc;
124         struct hlist_node list;
125         struct connection *othercon;
126         struct work_struct rwork; /* Receive workqueue */
127         struct work_struct swork; /* Send workqueue */
128 };
129 #define sock2con(x) ((struct connection *)(x)->sk_user_data)
130
131 /* An entry waiting to be sent */
132 struct writequeue_entry {
133         struct list_head list;
134         struct page *page;
135         int offset;
136         int len;
137         int end;
138         int users;
139         struct connection *con;
140 };
141
142 static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
143 static int dlm_local_count;
144
145 /* Work queues */
146 static struct workqueue_struct *recv_workqueue;
147 static struct workqueue_struct *send_workqueue;
148
149 static struct hlist_head connection_hash[CONN_HASH_SIZE];
150 static DEFINE_MUTEX(connections_lock);
151 static struct kmem_cache *con_cache;
152
153 static void process_recv_sockets(struct work_struct *work);
154 static void process_send_sockets(struct work_struct *work);
155
156
157 /* This is deliberately very simple because most clusters have simple
158    sequential nodeids, so we should be able to go straight to a connection
159    struct in the array */
160 static inline int nodeid_hash(int nodeid)
161 {
162         return nodeid & (CONN_HASH_SIZE-1);
163 }
164
165 static struct connection *__find_con(int nodeid)
166 {
167         int r;
168         struct hlist_node *h;
169         struct connection *con;
170
171         r = nodeid_hash(nodeid);
172
173         hlist_for_each_entry(con, h, &connection_hash[r], list) {
174                 if (con->nodeid == nodeid)
175                         return con;
176         }
177         return NULL;
178 }
179
180 /*
181  * If 'allocation' is zero then we don't attempt to create a new
182  * connection structure for this node.
183  */
184 static struct connection *__nodeid2con(int nodeid, gfp_t alloc)
185 {
186         struct connection *con = NULL;
187         int r;
188
189         con = __find_con(nodeid);
190         if (con || !alloc)
191                 return con;
192
193         con = kmem_cache_zalloc(con_cache, alloc);
194         if (!con)
195                 return NULL;
196
197         r = nodeid_hash(nodeid);
198         hlist_add_head(&con->list, &connection_hash[r]);
199
200         con->nodeid = nodeid;
201         mutex_init(&con->sock_mutex);
202         INIT_LIST_HEAD(&con->writequeue);
203         spin_lock_init(&con->writequeue_lock);
204         INIT_WORK(&con->swork, process_send_sockets);
205         INIT_WORK(&con->rwork, process_recv_sockets);
206
207         /* Setup action pointers for child sockets */
208         if (con->nodeid) {
209                 struct connection *zerocon = __find_con(0);
210
211                 con->connect_action = zerocon->connect_action;
212                 if (!con->rx_action)
213                         con->rx_action = zerocon->rx_action;
214         }
215
216         return con;
217 }
218
219 /* Loop round all connections */
220 static void foreach_conn(void (*conn_func)(struct connection *c))
221 {
222         int i;
223         struct hlist_node *h, *n;
224         struct connection *con;
225
226         for (i = 0; i < CONN_HASH_SIZE; i++) {
227                 hlist_for_each_entry_safe(con, h, n, &connection_hash[i], list){
228                         conn_func(con);
229                 }
230         }
231 }
232
233 static struct connection *nodeid2con(int nodeid, gfp_t allocation)
234 {
235         struct connection *con;
236
237         mutex_lock(&connections_lock);
238         con = __nodeid2con(nodeid, allocation);
239         mutex_unlock(&connections_lock);
240
241         return con;
242 }
243
244 /* This is a bit drastic, but only called when things go wrong */
245 static struct connection *assoc2con(int assoc_id)
246 {
247         int i;
248         struct hlist_node *h;
249         struct connection *con;
250
251         mutex_lock(&connections_lock);
252
253         for (i = 0 ; i < CONN_HASH_SIZE; i++) {
254                 hlist_for_each_entry(con, h, &connection_hash[i], list) {
255                         if (con->sctp_assoc == assoc_id) {
256                                 mutex_unlock(&connections_lock);
257                                 return con;
258                         }
259                 }
260         }
261         mutex_unlock(&connections_lock);
262         return NULL;
263 }
264
265 static int nodeid_to_addr(int nodeid, struct sockaddr *retaddr)
266 {
267         struct sockaddr_storage addr;
268         int error;
269
270         if (!dlm_local_count)
271                 return -1;
272
273         error = dlm_nodeid_to_addr(nodeid, &addr);
274         if (error)
275                 return error;
276
277         if (dlm_local_addr[0]->ss_family == AF_INET) {
278                 struct sockaddr_in *in4  = (struct sockaddr_in *) &addr;
279                 struct sockaddr_in *ret4 = (struct sockaddr_in *) retaddr;
280                 ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
281         } else {
282                 struct sockaddr_in6 *in6  = (struct sockaddr_in6 *) &addr;
283                 struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) retaddr;
284                 ipv6_addr_copy(&ret6->sin6_addr, &in6->sin6_addr);
285         }
286
287         return 0;
288 }
289
290 /* Data available on socket or listen socket received a connect */
291 static void lowcomms_data_ready(struct sock *sk, int count_unused)
292 {
293         struct connection *con = sock2con(sk);
294         if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
295                 queue_work(recv_workqueue, &con->rwork);
296 }
297
298 static void lowcomms_write_space(struct sock *sk)
299 {
300         struct connection *con = sock2con(sk);
301
302         if (!con)
303                 return;
304
305         clear_bit(SOCK_NOSPACE, &con->sock->flags);
306
307         if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
308                 con->sock->sk->sk_write_pending--;
309                 clear_bit(SOCK_ASYNC_NOSPACE, &con->sock->flags);
310         }
311
312         if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
313                 queue_work(send_workqueue, &con->swork);
314 }
315
316 static inline void lowcomms_connect_sock(struct connection *con)
317 {
318         if (test_bit(CF_CLOSE, &con->flags))
319                 return;
320         if (!test_and_set_bit(CF_CONNECT_PENDING, &con->flags))
321                 queue_work(send_workqueue, &con->swork);
322 }
323
324 static void lowcomms_state_change(struct sock *sk)
325 {
326         if (sk->sk_state == TCP_ESTABLISHED)
327                 lowcomms_write_space(sk);
328 }
329
330 int dlm_lowcomms_connect_node(int nodeid)
331 {
332         struct connection *con;
333
334         /* with sctp there's no connecting without sending */
335         if (dlm_config.ci_protocol != 0)
336                 return 0;
337
338         if (nodeid == dlm_our_nodeid())
339                 return 0;
340
341         con = nodeid2con(nodeid, GFP_NOFS);
342         if (!con)
343                 return -ENOMEM;
344         lowcomms_connect_sock(con);
345         return 0;
346 }
347
348 /* Make a socket active */
349 static int add_sock(struct socket *sock, struct connection *con)
350 {
351         con->sock = sock;
352
353         /* Install a data_ready callback */
354         con->sock->sk->sk_data_ready = lowcomms_data_ready;
355         con->sock->sk->sk_write_space = lowcomms_write_space;
356         con->sock->sk->sk_state_change = lowcomms_state_change;
357         con->sock->sk->sk_user_data = con;
358         con->sock->sk->sk_allocation = GFP_NOFS;
359         return 0;
360 }
361
362 /* Add the port number to an IPv6 or 4 sockaddr and return the address
363    length */
364 static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
365                           int *addr_len)
366 {
367         saddr->ss_family =  dlm_local_addr[0]->ss_family;
368         if (saddr->ss_family == AF_INET) {
369                 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
370                 in4_addr->sin_port = cpu_to_be16(port);
371                 *addr_len = sizeof(struct sockaddr_in);
372                 memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
373         } else {
374                 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
375                 in6_addr->sin6_port = cpu_to_be16(port);
376                 *addr_len = sizeof(struct sockaddr_in6);
377         }
378         memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
379 }
380
381 /* Close a remote connection and tidy up */
382 static void close_connection(struct connection *con, bool and_other)
383 {
384         mutex_lock(&con->sock_mutex);
385
386         if (con->sock) {
387                 sock_release(con->sock);
388                 con->sock = NULL;
389         }
390         if (con->othercon && and_other) {
391                 /* Will only re-enter once. */
392                 close_connection(con->othercon, false);
393         }
394         if (con->rx_page) {
395                 __free_page(con->rx_page);
396                 con->rx_page = NULL;
397         }
398
399         con->retries = 0;
400         mutex_unlock(&con->sock_mutex);
401 }
402
403 /* We only send shutdown messages to nodes that are not part of the cluster */
404 static void sctp_send_shutdown(sctp_assoc_t associd)
405 {
406         static char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
407         struct msghdr outmessage;
408         struct cmsghdr *cmsg;
409         struct sctp_sndrcvinfo *sinfo;
410         int ret;
411         struct connection *con;
412
413         con = nodeid2con(0,0);
414         BUG_ON(con == NULL);
415
416         outmessage.msg_name = NULL;
417         outmessage.msg_namelen = 0;
418         outmessage.msg_control = outcmsg;
419         outmessage.msg_controllen = sizeof(outcmsg);
420         outmessage.msg_flags = MSG_EOR;
421
422         cmsg = CMSG_FIRSTHDR(&outmessage);
423         cmsg->cmsg_level = IPPROTO_SCTP;
424         cmsg->cmsg_type = SCTP_SNDRCV;
425         cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
426         outmessage.msg_controllen = cmsg->cmsg_len;
427         sinfo = CMSG_DATA(cmsg);
428         memset(sinfo, 0x00, sizeof(struct sctp_sndrcvinfo));
429
430         sinfo->sinfo_flags |= MSG_EOF;
431         sinfo->sinfo_assoc_id = associd;
432
433         ret = kernel_sendmsg(con->sock, &outmessage, NULL, 0, 0);
434
435         if (ret != 0)
436                 log_print("send EOF to node failed: %d", ret);
437 }
438
439 static void sctp_init_failed_foreach(struct connection *con)
440 {
441         con->sctp_assoc = 0;
442         if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
443                 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
444                         queue_work(send_workqueue, &con->swork);
445         }
446 }
447
448 /* INIT failed but we don't know which node...
449    restart INIT on all pending nodes */
450 static void sctp_init_failed(void)
451 {
452         mutex_lock(&connections_lock);
453
454         foreach_conn(sctp_init_failed_foreach);
455
456         mutex_unlock(&connections_lock);
457 }
458
459 /* Something happened to an association */
460 static void process_sctp_notification(struct connection *con,
461                                       struct msghdr *msg, char *buf)
462 {
463         union sctp_notification *sn = (union sctp_notification *)buf;
464
465         if (sn->sn_header.sn_type == SCTP_ASSOC_CHANGE) {
466                 switch (sn->sn_assoc_change.sac_state) {
467
468                 case SCTP_COMM_UP:
469                 case SCTP_RESTART:
470                 {
471                         /* Check that the new node is in the lockspace */
472                         struct sctp_prim prim;
473                         int nodeid;
474                         int prim_len, ret;
475                         int addr_len;
476                         struct connection *new_con;
477                         sctp_peeloff_arg_t parg;
478                         int parglen = sizeof(parg);
479                         int err;
480
481                         /*
482                          * We get this before any data for an association.
483                          * We verify that the node is in the cluster and
484                          * then peel off a socket for it.
485                          */
486                         if ((int)sn->sn_assoc_change.sac_assoc_id <= 0) {
487                                 log_print("COMM_UP for invalid assoc ID %d",
488                                          (int)sn->sn_assoc_change.sac_assoc_id);
489                                 sctp_init_failed();
490                                 return;
491                         }
492                         memset(&prim, 0, sizeof(struct sctp_prim));
493                         prim_len = sizeof(struct sctp_prim);
494                         prim.ssp_assoc_id = sn->sn_assoc_change.sac_assoc_id;
495
496                         ret = kernel_getsockopt(con->sock,
497                                                 IPPROTO_SCTP,
498                                                 SCTP_PRIMARY_ADDR,
499                                                 (char*)&prim,
500                                                 &prim_len);
501                         if (ret < 0) {
502                                 log_print("getsockopt/sctp_primary_addr on "
503                                           "new assoc %d failed : %d",
504                                           (int)sn->sn_assoc_change.sac_assoc_id,
505                                           ret);
506
507                                 /* Retry INIT later */
508                                 new_con = assoc2con(sn->sn_assoc_change.sac_assoc_id);
509                                 if (new_con)
510                                         clear_bit(CF_CONNECT_PENDING, &con->flags);
511                                 return;
512                         }
513                         make_sockaddr(&prim.ssp_addr, 0, &addr_len);
514                         if (dlm_addr_to_nodeid(&prim.ssp_addr, &nodeid)) {
515                                 int i;
516                                 unsigned char *b=(unsigned char *)&prim.ssp_addr;
517                                 log_print("reject connect from unknown addr");
518                                 for (i=0; i<sizeof(struct sockaddr_storage);i++)
519                                         printk("%02x ", b[i]);
520                                 printk("\n");
521                                 sctp_send_shutdown(prim.ssp_assoc_id);
522                                 return;
523                         }
524
525                         new_con = nodeid2con(nodeid, GFP_NOFS);
526                         if (!new_con)
527                                 return;
528
529                         /* Peel off a new sock */
530                         parg.associd = sn->sn_assoc_change.sac_assoc_id;
531                         ret = kernel_getsockopt(con->sock, IPPROTO_SCTP,
532                                                 SCTP_SOCKOPT_PEELOFF,
533                                                 (void *)&parg, &parglen);
534                         if (ret < 0) {
535                                 log_print("Can't peel off a socket for "
536                                           "connection %d to node %d: err=%d",
537                                           parg.associd, nodeid, ret);
538                                 return;
539                         }
540                         new_con->sock = sockfd_lookup(parg.sd, &err);
541                         if (!new_con->sock) {
542                                 log_print("sockfd_lookup error %d", err);
543                                 return;
544                         }
545                         add_sock(new_con->sock, new_con);
546                         sockfd_put(new_con->sock);
547
548                         log_print("connecting to %d sctp association %d",
549                                  nodeid, (int)sn->sn_assoc_change.sac_assoc_id);
550
551                         /* Send any pending writes */
552                         clear_bit(CF_CONNECT_PENDING, &new_con->flags);
553                         clear_bit(CF_INIT_PENDING, &con->flags);
554                         if (!test_and_set_bit(CF_WRITE_PENDING, &new_con->flags)) {
555                                 queue_work(send_workqueue, &new_con->swork);
556                         }
557                         if (!test_and_set_bit(CF_READ_PENDING, &new_con->flags))
558                                 queue_work(recv_workqueue, &new_con->rwork);
559                 }
560                 break;
561
562                 case SCTP_COMM_LOST:
563                 case SCTP_SHUTDOWN_COMP:
564                 {
565                         con = assoc2con(sn->sn_assoc_change.sac_assoc_id);
566                         if (con) {
567                                 con->sctp_assoc = 0;
568                         }
569                 }
570                 break;
571
572                 /* We don't know which INIT failed, so clear the PENDING flags
573                  * on them all.  if assoc_id is zero then it will then try
574                  * again */
575
576                 case SCTP_CANT_STR_ASSOC:
577                 {
578                         log_print("Can't start SCTP association - retrying");
579                         sctp_init_failed();
580                 }
581                 break;
582
583                 default:
584                         log_print("unexpected SCTP assoc change id=%d state=%d",
585                                   (int)sn->sn_assoc_change.sac_assoc_id,
586                                   sn->sn_assoc_change.sac_state);
587                 }
588         }
589 }
590
591 /* Data received from remote end */
592 static int receive_from_sock(struct connection *con)
593 {
594         int ret = 0;
595         struct msghdr msg = {};
596         struct kvec iov[2];
597         unsigned len;
598         int r;
599         int call_again_soon = 0;
600         int nvec;
601         char incmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
602
603         mutex_lock(&con->sock_mutex);
604
605         if (con->sock == NULL) {
606                 ret = -EAGAIN;
607                 goto out_close;
608         }
609
610         if (con->rx_page == NULL) {
611                 /*
612                  * This doesn't need to be atomic, but I think it should
613                  * improve performance if it is.
614                  */
615                 con->rx_page = alloc_page(GFP_ATOMIC);
616                 if (con->rx_page == NULL)
617                         goto out_resched;
618                 cbuf_init(&con->cb, PAGE_CACHE_SIZE);
619         }
620
621         /* Only SCTP needs these really */
622         memset(&incmsg, 0, sizeof(incmsg));
623         msg.msg_control = incmsg;
624         msg.msg_controllen = sizeof(incmsg);
625
626         /*
627          * iov[0] is the bit of the circular buffer between the current end
628          * point (cb.base + cb.len) and the end of the buffer.
629          */
630         iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
631         iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
632         iov[1].iov_len = 0;
633         nvec = 1;
634
635         /*
636          * iov[1] is the bit of the circular buffer between the start of the
637          * buffer and the start of the currently used section (cb.base)
638          */
639         if (cbuf_data(&con->cb) >= con->cb.base) {
640                 iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);
641                 iov[1].iov_len = con->cb.base;
642                 iov[1].iov_base = page_address(con->rx_page);
643                 nvec = 2;
644         }
645         len = iov[0].iov_len + iov[1].iov_len;
646
647         r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,
648                                MSG_DONTWAIT | MSG_NOSIGNAL);
649         if (ret <= 0)
650                 goto out_close;
651
652         /* Process SCTP notifications */
653         if (msg.msg_flags & MSG_NOTIFICATION) {
654                 msg.msg_control = incmsg;
655                 msg.msg_controllen = sizeof(incmsg);
656
657                 process_sctp_notification(con, &msg,
658                                 page_address(con->rx_page) + con->cb.base);
659                 mutex_unlock(&con->sock_mutex);
660                 return 0;
661         }
662         BUG_ON(con->nodeid == 0);
663
664         if (ret == len)
665                 call_again_soon = 1;
666         cbuf_add(&con->cb, ret);
667         ret = dlm_process_incoming_buffer(con->nodeid,
668                                           page_address(con->rx_page),
669                                           con->cb.base, con->cb.len,
670                                           PAGE_CACHE_SIZE);
671         if (ret == -EBADMSG) {
672                 log_print("lowcomms: addr=%p, base=%u, len=%u, "
673                           "iov_len=%u, iov_base[0]=%p, read=%d",
674                           page_address(con->rx_page), con->cb.base, con->cb.len,
675                           len, iov[0].iov_base, r);
676         }
677         if (ret < 0)
678                 goto out_close;
679         cbuf_eat(&con->cb, ret);
680
681         if (cbuf_empty(&con->cb) && !call_again_soon) {
682                 __free_page(con->rx_page);
683                 con->rx_page = NULL;
684         }
685
686         if (call_again_soon)
687                 goto out_resched;
688         mutex_unlock(&con->sock_mutex);
689         return 0;
690
691 out_resched:
692         if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
693                 queue_work(recv_workqueue, &con->rwork);
694         mutex_unlock(&con->sock_mutex);
695         return -EAGAIN;
696
697 out_close:
698         mutex_unlock(&con->sock_mutex);
699         if (ret != -EAGAIN) {
700                 close_connection(con, false);
701                 /* Reconnect when there is something to send */
702         }
703         /* Don't return success if we really got EOF */
704         if (ret == 0)
705                 ret = -EAGAIN;
706
707         return ret;
708 }
709
710 /* Listening socket is busy, accept a connection */
711 static int tcp_accept_from_sock(struct connection *con)
712 {
713         int result;
714         struct sockaddr_storage peeraddr;
715         struct socket *newsock;
716         int len;
717         int nodeid;
718         struct connection *newcon;
719         struct connection *addcon;
720
721         memset(&peeraddr, 0, sizeof(peeraddr));
722         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
723                                   IPPROTO_TCP, &newsock);
724         if (result < 0)
725                 return -ENOMEM;
726
727         mutex_lock_nested(&con->sock_mutex, 0);
728
729         result = -ENOTCONN;
730         if (con->sock == NULL)
731                 goto accept_err;
732
733         newsock->type = con->sock->type;
734         newsock->ops = con->sock->ops;
735
736         result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK);
737         if (result < 0)
738                 goto accept_err;
739
740         /* Get the connected socket's peer */
741         memset(&peeraddr, 0, sizeof(peeraddr));
742         if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr,
743                                   &len, 2)) {
744                 result = -ECONNABORTED;
745                 goto accept_err;
746         }
747
748         /* Get the new node's NODEID */
749         make_sockaddr(&peeraddr, 0, &len);
750         if (dlm_addr_to_nodeid(&peeraddr, &nodeid)) {
751                 log_print("connect from non cluster node");
752                 sock_release(newsock);
753                 mutex_unlock(&con->sock_mutex);
754                 return -1;
755         }
756
757         log_print("got connection from %d", nodeid);
758
759         /*  Check to see if we already have a connection to this node. This
760          *  could happen if the two nodes initiate a connection at roughly
761          *  the same time and the connections cross on the wire.
762          *  In this case we store the incoming one in "othercon"
763          */
764         newcon = nodeid2con(nodeid, GFP_NOFS);
765         if (!newcon) {
766                 result = -ENOMEM;
767                 goto accept_err;
768         }
769         mutex_lock_nested(&newcon->sock_mutex, 1);
770         if (newcon->sock) {
771                 struct connection *othercon = newcon->othercon;
772
773                 if (!othercon) {
774                         othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
775                         if (!othercon) {
776                                 log_print("failed to allocate incoming socket");
777                                 mutex_unlock(&newcon->sock_mutex);
778                                 result = -ENOMEM;
779                                 goto accept_err;
780                         }
781                         othercon->nodeid = nodeid;
782                         othercon->rx_action = receive_from_sock;
783                         mutex_init(&othercon->sock_mutex);
784                         INIT_WORK(&othercon->swork, process_send_sockets);
785                         INIT_WORK(&othercon->rwork, process_recv_sockets);
786                         set_bit(CF_IS_OTHERCON, &othercon->flags);
787                 }
788                 if (!othercon->sock) {
789                         newcon->othercon = othercon;
790                         othercon->sock = newsock;
791                         newsock->sk->sk_user_data = othercon;
792                         add_sock(newsock, othercon);
793                         addcon = othercon;
794                 }
795                 else {
796                         printk("Extra connection from node %d attempted\n", nodeid);
797                         result = -EAGAIN;
798                         mutex_unlock(&newcon->sock_mutex);
799                         goto accept_err;
800                 }
801         }
802         else {
803                 newsock->sk->sk_user_data = newcon;
804                 newcon->rx_action = receive_from_sock;
805                 add_sock(newsock, newcon);
806                 addcon = newcon;
807         }
808
809         mutex_unlock(&newcon->sock_mutex);
810
811         /*
812          * Add it to the active queue in case we got data
813          * beween processing the accept adding the socket
814          * to the read_sockets list
815          */
816         if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
817                 queue_work(recv_workqueue, &addcon->rwork);
818         mutex_unlock(&con->sock_mutex);
819
820         return 0;
821
822 accept_err:
823         mutex_unlock(&con->sock_mutex);
824         sock_release(newsock);
825
826         if (result != -EAGAIN)
827                 log_print("error accepting connection from node: %d", result);
828         return result;
829 }
830
831 static void free_entry(struct writequeue_entry *e)
832 {
833         __free_page(e->page);
834         kfree(e);
835 }
836
837 /* Initiate an SCTP association.
838    This is a special case of send_to_sock() in that we don't yet have a
839    peeled-off socket for this association, so we use the listening socket
840    and add the primary IP address of the remote node.
841  */
842 static void sctp_init_assoc(struct connection *con)
843 {
844         struct sockaddr_storage rem_addr;
845         char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
846         struct msghdr outmessage;
847         struct cmsghdr *cmsg;
848         struct sctp_sndrcvinfo *sinfo;
849         struct connection *base_con;
850         struct writequeue_entry *e;
851         int len, offset;
852         int ret;
853         int addrlen;
854         struct kvec iov[1];
855
856         if (test_and_set_bit(CF_INIT_PENDING, &con->flags))
857                 return;
858
859         if (con->retries++ > MAX_CONNECT_RETRIES)
860                 return;
861
862         if (nodeid_to_addr(con->nodeid, (struct sockaddr *)&rem_addr)) {
863                 log_print("no address for nodeid %d", con->nodeid);
864                 return;
865         }
866         base_con = nodeid2con(0, 0);
867         BUG_ON(base_con == NULL);
868
869         make_sockaddr(&rem_addr, dlm_config.ci_tcp_port, &addrlen);
870
871         outmessage.msg_name = &rem_addr;
872         outmessage.msg_namelen = addrlen;
873         outmessage.msg_control = outcmsg;
874         outmessage.msg_controllen = sizeof(outcmsg);
875         outmessage.msg_flags = MSG_EOR;
876
877         spin_lock(&con->writequeue_lock);
878
879         if (list_empty(&con->writequeue)) {
880                 spin_unlock(&con->writequeue_lock);
881                 log_print("writequeue empty for nodeid %d", con->nodeid);
882                 return;
883         }
884
885         e = list_first_entry(&con->writequeue, struct writequeue_entry, list);
886         len = e->len;
887         offset = e->offset;
888         spin_unlock(&con->writequeue_lock);
889
890         /* Send the first block off the write queue */
891         iov[0].iov_base = page_address(e->page)+offset;
892         iov[0].iov_len = len;
893
894         cmsg = CMSG_FIRSTHDR(&outmessage);
895         cmsg->cmsg_level = IPPROTO_SCTP;
896         cmsg->cmsg_type = SCTP_SNDRCV;
897         cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
898         sinfo = CMSG_DATA(cmsg);
899         memset(sinfo, 0x00, sizeof(struct sctp_sndrcvinfo));
900         sinfo->sinfo_ppid = cpu_to_le32(dlm_our_nodeid());
901         outmessage.msg_controllen = cmsg->cmsg_len;
902
903         ret = kernel_sendmsg(base_con->sock, &outmessage, iov, 1, len);
904         if (ret < 0) {
905                 log_print("Send first packet to node %d failed: %d",
906                           con->nodeid, ret);
907
908                 /* Try again later */
909                 clear_bit(CF_CONNECT_PENDING, &con->flags);
910                 clear_bit(CF_INIT_PENDING, &con->flags);
911         }
912         else {
913                 spin_lock(&con->writequeue_lock);
914                 e->offset += ret;
915                 e->len -= ret;
916
917                 if (e->len == 0 && e->users == 0) {
918                         list_del(&e->list);
919                         free_entry(e);
920                 }
921                 spin_unlock(&con->writequeue_lock);
922         }
923 }
924
925 /* Connect a new socket to its peer */
926 static void tcp_connect_to_sock(struct connection *con)
927 {
928         int result = -EHOSTUNREACH;
929         struct sockaddr_storage saddr, src_addr;
930         int addr_len;
931         struct socket *sock = NULL;
932         int one = 1;
933
934         if (con->nodeid == 0) {
935                 log_print("attempt to connect sock 0 foiled");
936                 return;
937         }
938
939         mutex_lock(&con->sock_mutex);
940         if (con->retries++ > MAX_CONNECT_RETRIES)
941                 goto out;
942
943         /* Some odd races can cause double-connects, ignore them */
944         if (con->sock) {
945                 result = 0;
946                 goto out;
947         }
948
949         /* Create a socket to communicate with */
950         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
951                                   IPPROTO_TCP, &sock);
952         if (result < 0)
953                 goto out_err;
954
955         memset(&saddr, 0, sizeof(saddr));
956         if (dlm_nodeid_to_addr(con->nodeid, &saddr))
957                 goto out_err;
958
959         sock->sk->sk_user_data = con;
960         con->rx_action = receive_from_sock;
961         con->connect_action = tcp_connect_to_sock;
962         add_sock(sock, con);
963
964         /* Bind to our cluster-known address connecting to avoid
965            routing problems */
966         memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
967         make_sockaddr(&src_addr, 0, &addr_len);
968         result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
969                                  addr_len);
970         if (result < 0) {
971                 log_print("could not bind for connect: %d", result);
972                 /* This *may* not indicate a critical error */
973         }
974
975         make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
976
977         log_print("connecting to %d", con->nodeid);
978
979         /* Turn off Nagle's algorithm */
980         kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
981                           sizeof(one));
982
983         result =
984                 sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
985                                    O_NONBLOCK);
986         if (result == -EINPROGRESS)
987                 result = 0;
988         if (result == 0)
989                 goto out;
990
991 out_err:
992         if (con->sock) {
993                 sock_release(con->sock);
994                 con->sock = NULL;
995         } else if (sock) {
996                 sock_release(sock);
997         }
998         /*
999          * Some errors are fatal and this list might need adjusting. For other
1000          * errors we try again until the max number of retries is reached.
1001          */
1002         if (result != -EHOSTUNREACH && result != -ENETUNREACH &&
1003             result != -ENETDOWN && result != -EINVAL
1004             && result != -EPROTONOSUPPORT) {
1005                 lowcomms_connect_sock(con);
1006                 result = 0;
1007         }
1008 out:
1009         mutex_unlock(&con->sock_mutex);
1010         return;
1011 }
1012
1013 static struct socket *tcp_create_listen_sock(struct connection *con,
1014                                              struct sockaddr_storage *saddr)
1015 {
1016         struct socket *sock = NULL;
1017         int result = 0;
1018         int one = 1;
1019         int addr_len;
1020
1021         if (dlm_local_addr[0]->ss_family == AF_INET)
1022                 addr_len = sizeof(struct sockaddr_in);
1023         else
1024                 addr_len = sizeof(struct sockaddr_in6);
1025
1026         /* Create a socket to communicate with */
1027         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
1028                                   IPPROTO_TCP, &sock);
1029         if (result < 0) {
1030                 log_print("Can't create listening comms socket");
1031                 goto create_out;
1032         }
1033
1034         /* Turn off Nagle's algorithm */
1035         kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1036                           sizeof(one));
1037
1038         result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
1039                                    (char *)&one, sizeof(one));
1040
1041         if (result < 0) {
1042                 log_print("Failed to set SO_REUSEADDR on socket: %d", result);
1043         }
1044         sock->sk->sk_user_data = con;
1045         con->rx_action = tcp_accept_from_sock;
1046         con->connect_action = tcp_connect_to_sock;
1047         con->sock = sock;
1048
1049         /* Bind to our port */
1050         make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
1051         result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
1052         if (result < 0) {
1053                 log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
1054                 sock_release(sock);
1055                 sock = NULL;
1056                 con->sock = NULL;
1057                 goto create_out;
1058         }
1059         result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
1060                                  (char *)&one, sizeof(one));
1061         if (result < 0) {
1062                 log_print("Set keepalive failed: %d", result);
1063         }
1064
1065         result = sock->ops->listen(sock, 5);
1066         if (result < 0) {
1067                 log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
1068                 sock_release(sock);
1069                 sock = NULL;
1070                 goto create_out;
1071         }
1072
1073 create_out:
1074         return sock;
1075 }
1076
1077 /* Get local addresses */
1078 static void init_local(void)
1079 {
1080         struct sockaddr_storage sas, *addr;
1081         int i;
1082
1083         dlm_local_count = 0;
1084         for (i = 0; i < DLM_MAX_ADDR_COUNT - 1; i++) {
1085                 if (dlm_our_addr(&sas, i))
1086                         break;
1087
1088                 addr = kmalloc(sizeof(*addr), GFP_NOFS);
1089                 if (!addr)
1090                         break;
1091                 memcpy(addr, &sas, sizeof(*addr));
1092                 dlm_local_addr[dlm_local_count++] = addr;
1093         }
1094 }
1095
1096 /* Bind to an IP address. SCTP allows multiple address so it can do
1097    multi-homing */
1098 static int add_sctp_bind_addr(struct connection *sctp_con,
1099                               struct sockaddr_storage *addr,
1100                               int addr_len, int num)
1101 {
1102         int result = 0;
1103
1104         if (num == 1)
1105                 result = kernel_bind(sctp_con->sock,
1106                                      (struct sockaddr *) addr,
1107                                      addr_len);
1108         else
1109                 result = kernel_setsockopt(sctp_con->sock, SOL_SCTP,
1110                                            SCTP_SOCKOPT_BINDX_ADD,
1111                                            (char *)addr, addr_len);
1112
1113         if (result < 0)
1114                 log_print("Can't bind to port %d addr number %d",
1115                           dlm_config.ci_tcp_port, num);
1116
1117         return result;
1118 }
1119
1120 /* Initialise SCTP socket and bind to all interfaces */
1121 static int sctp_listen_for_all(void)
1122 {
1123         struct socket *sock = NULL;
1124         struct sockaddr_storage localaddr;
1125         struct sctp_event_subscribe subscribe;
1126         int result = -EINVAL, num = 1, i, addr_len;
1127         struct connection *con = nodeid2con(0, GFP_NOFS);
1128         int bufsize = NEEDED_RMEM;
1129
1130         if (!con)
1131                 return -ENOMEM;
1132
1133         log_print("Using SCTP for communications");
1134
1135         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_SEQPACKET,
1136                                   IPPROTO_SCTP, &sock);
1137         if (result < 0) {
1138                 log_print("Can't create comms socket, check SCTP is loaded");
1139                 goto out;
1140         }
1141
1142         /* Listen for events */
1143         memset(&subscribe, 0, sizeof(subscribe));
1144         subscribe.sctp_data_io_event = 1;
1145         subscribe.sctp_association_event = 1;
1146         subscribe.sctp_send_failure_event = 1;
1147         subscribe.sctp_shutdown_event = 1;
1148         subscribe.sctp_partial_delivery_event = 1;
1149
1150         result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
1151                                  (char *)&bufsize, sizeof(bufsize));
1152         if (result)
1153                 log_print("Error increasing buffer space on socket %d", result);
1154
1155         result = kernel_setsockopt(sock, SOL_SCTP, SCTP_EVENTS,
1156                                    (char *)&subscribe, sizeof(subscribe));
1157         if (result < 0) {
1158                 log_print("Failed to set SCTP_EVENTS on socket: result=%d",
1159                           result);
1160                 goto create_delsock;
1161         }
1162
1163         /* Init con struct */
1164         sock->sk->sk_user_data = con;
1165         con->sock = sock;
1166         con->sock->sk->sk_data_ready = lowcomms_data_ready;
1167         con->rx_action = receive_from_sock;
1168         con->connect_action = sctp_init_assoc;
1169
1170         /* Bind to all interfaces. */
1171         for (i = 0; i < dlm_local_count; i++) {
1172                 memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr));
1173                 make_sockaddr(&localaddr, dlm_config.ci_tcp_port, &addr_len);
1174
1175                 result = add_sctp_bind_addr(con, &localaddr, addr_len, num);
1176                 if (result)
1177                         goto create_delsock;
1178                 ++num;
1179         }
1180
1181         result = sock->ops->listen(sock, 5);
1182         if (result < 0) {
1183                 log_print("Can't set socket listening");
1184                 goto create_delsock;
1185         }
1186
1187         return 0;
1188
1189 create_delsock:
1190         sock_release(sock);
1191         con->sock = NULL;
1192 out:
1193         return result;
1194 }
1195
1196 static int tcp_listen_for_all(void)
1197 {
1198         struct socket *sock = NULL;
1199         struct connection *con = nodeid2con(0, GFP_NOFS);
1200         int result = -EINVAL;
1201
1202         if (!con)
1203                 return -ENOMEM;
1204
1205         /* We don't support multi-homed hosts */
1206         if (dlm_local_addr[1] != NULL) {
1207                 log_print("TCP protocol can't handle multi-homed hosts, "
1208                           "try SCTP");
1209                 return -EINVAL;
1210         }
1211
1212         log_print("Using TCP for communications");
1213
1214         sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
1215         if (sock) {
1216                 add_sock(sock, con);
1217                 result = 0;
1218         }
1219         else {
1220                 result = -EADDRINUSE;
1221         }
1222
1223         return result;
1224 }
1225
1226
1227
1228 static struct writequeue_entry *new_writequeue_entry(struct connection *con,
1229                                                      gfp_t allocation)
1230 {
1231         struct writequeue_entry *entry;
1232
1233         entry = kmalloc(sizeof(struct writequeue_entry), allocation);
1234         if (!entry)
1235                 return NULL;
1236
1237         entry->page = alloc_page(allocation);
1238         if (!entry->page) {
1239                 kfree(entry);
1240                 return NULL;
1241         }
1242
1243         entry->offset = 0;
1244         entry->len = 0;
1245         entry->end = 0;
1246         entry->users = 0;
1247         entry->con = con;
1248
1249         return entry;
1250 }
1251
1252 void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
1253 {
1254         struct connection *con;
1255         struct writequeue_entry *e;
1256         int offset = 0;
1257         int users = 0;
1258
1259         con = nodeid2con(nodeid, allocation);
1260         if (!con)
1261                 return NULL;
1262
1263         spin_lock(&con->writequeue_lock);
1264         e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
1265         if ((&e->list == &con->writequeue) ||
1266             (PAGE_CACHE_SIZE - e->end < len)) {
1267                 e = NULL;
1268         } else {
1269                 offset = e->end;
1270                 e->end += len;
1271                 users = e->users++;
1272         }
1273         spin_unlock(&con->writequeue_lock);
1274
1275         if (e) {
1276         got_one:
1277                 *ppc = page_address(e->page) + offset;
1278                 return e;
1279         }
1280
1281         e = new_writequeue_entry(con, allocation);
1282         if (e) {
1283                 spin_lock(&con->writequeue_lock);
1284                 offset = e->end;
1285                 e->end += len;
1286                 users = e->users++;
1287                 list_add_tail(&e->list, &con->writequeue);
1288                 spin_unlock(&con->writequeue_lock);
1289                 goto got_one;
1290         }
1291         return NULL;
1292 }
1293
1294 void dlm_lowcomms_commit_buffer(void *mh)
1295 {
1296         struct writequeue_entry *e = (struct writequeue_entry *)mh;
1297         struct connection *con = e->con;
1298         int users;
1299
1300         spin_lock(&con->writequeue_lock);
1301         users = --e->users;
1302         if (users)
1303                 goto out;
1304         e->len = e->end - e->offset;
1305         spin_unlock(&con->writequeue_lock);
1306
1307         if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {
1308                 queue_work(send_workqueue, &con->swork);
1309         }
1310         return;
1311
1312 out:
1313         spin_unlock(&con->writequeue_lock);
1314         return;
1315 }
1316
1317 /* Send a message */
1318 static void send_to_sock(struct connection *con)
1319 {
1320         int ret = 0;
1321         const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
1322         struct writequeue_entry *e;
1323         int len, offset;
1324         int count = 0;
1325
1326         mutex_lock(&con->sock_mutex);
1327         if (con->sock == NULL)
1328                 goto out_connect;
1329
1330         spin_lock(&con->writequeue_lock);
1331         for (;;) {
1332                 e = list_entry(con->writequeue.next, struct writequeue_entry,
1333                                list);
1334                 if ((struct list_head *) e == &con->writequeue)
1335                         break;
1336
1337                 len = e->len;
1338                 offset = e->offset;
1339                 BUG_ON(len == 0 && e->users == 0);
1340                 spin_unlock(&con->writequeue_lock);
1341
1342                 ret = 0;
1343                 if (len) {
1344                         ret = kernel_sendpage(con->sock, e->page, offset, len,
1345                                               msg_flags);
1346                         if (ret == -EAGAIN || ret == 0) {
1347                                 if (ret == -EAGAIN &&
1348                                     test_bit(SOCK_ASYNC_NOSPACE, &con->sock->flags) &&
1349                                     !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1350                                         /* Notify TCP that we're limited by the
1351                                          * application window size.
1352                                          */
1353                                         set_bit(SOCK_NOSPACE, &con->sock->flags);
1354                                         con->sock->sk->sk_write_pending++;
1355                                 }
1356                                 cond_resched();
1357                                 goto out;
1358                         }
1359                         if (ret <= 0)
1360                                 goto send_error;
1361                 }
1362
1363                 /* Don't starve people filling buffers */
1364                 if (++count >= MAX_SEND_MSG_COUNT) {
1365                         cond_resched();
1366                         count = 0;
1367                 }
1368
1369                 spin_lock(&con->writequeue_lock);
1370                 e->offset += ret;
1371                 e->len -= ret;
1372
1373                 if (e->len == 0 && e->users == 0) {
1374                         list_del(&e->list);
1375                         free_entry(e);
1376                         continue;
1377                 }
1378         }
1379         spin_unlock(&con->writequeue_lock);
1380 out:
1381         mutex_unlock(&con->sock_mutex);
1382         return;
1383
1384 send_error:
1385         mutex_unlock(&con->sock_mutex);
1386         close_connection(con, false);
1387         lowcomms_connect_sock(con);
1388         return;
1389
1390 out_connect:
1391         mutex_unlock(&con->sock_mutex);
1392         if (!test_bit(CF_INIT_PENDING, &con->flags))
1393                 lowcomms_connect_sock(con);
1394         return;
1395 }
1396
1397 static void clean_one_writequeue(struct connection *con)
1398 {
1399         struct writequeue_entry *e, *safe;
1400
1401         spin_lock(&con->writequeue_lock);
1402         list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1403                 list_del(&e->list);
1404                 free_entry(e);
1405         }
1406         spin_unlock(&con->writequeue_lock);
1407 }
1408
1409 /* Called from recovery when it knows that a node has
1410    left the cluster */
1411 int dlm_lowcomms_close(int nodeid)
1412 {
1413         struct connection *con;
1414
1415         log_print("closing connection to node %d", nodeid);
1416         con = nodeid2con(nodeid, 0);
1417         if (con) {
1418                 clear_bit(CF_CONNECT_PENDING, &con->flags);
1419                 clear_bit(CF_WRITE_PENDING, &con->flags);
1420                 set_bit(CF_CLOSE, &con->flags);
1421                 if (cancel_work_sync(&con->swork))
1422                         log_print("canceled swork for node %d", nodeid);
1423                 if (cancel_work_sync(&con->rwork))
1424                         log_print("canceled rwork for node %d", nodeid);
1425                 clean_one_writequeue(con);
1426                 close_connection(con, true);
1427         }
1428         return 0;
1429 }
1430
1431 /* Receive workqueue function */
1432 static void process_recv_sockets(struct work_struct *work)
1433 {
1434         struct connection *con = container_of(work, struct connection, rwork);
1435         int err;
1436
1437         clear_bit(CF_READ_PENDING, &con->flags);
1438         do {
1439                 err = con->rx_action(con);
1440         } while (!err);
1441 }
1442
1443 /* Send workqueue function */
1444 static void process_send_sockets(struct work_struct *work)
1445 {
1446         struct connection *con = container_of(work, struct connection, swork);
1447
1448         if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
1449                 con->connect_action(con);
1450                 set_bit(CF_WRITE_PENDING, &con->flags);
1451         }
1452         if (test_and_clear_bit(CF_WRITE_PENDING, &con->flags))
1453                 send_to_sock(con);
1454 }
1455
1456
1457 /* Discard all entries on the write queues */
1458 static void clean_writequeues(void)
1459 {
1460         foreach_conn(clean_one_writequeue);
1461 }
1462
1463 static void work_stop(void)
1464 {
1465         destroy_workqueue(recv_workqueue);
1466         destroy_workqueue(send_workqueue);
1467 }
1468
1469 static int work_start(void)
1470 {
1471         recv_workqueue = alloc_workqueue("dlm_recv",
1472                                          WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1473         if (!recv_workqueue) {
1474                 log_print("can't start dlm_recv");
1475                 return -ENOMEM;
1476         }
1477
1478         send_workqueue = alloc_workqueue("dlm_send",
1479                                          WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1480         if (!send_workqueue) {
1481                 log_print("can't start dlm_send");
1482                 destroy_workqueue(recv_workqueue);
1483                 return -ENOMEM;
1484         }
1485
1486         return 0;
1487 }
1488
1489 static void stop_conn(struct connection *con)
1490 {
1491         con->flags |= 0x0F;
1492         if (con->sock && con->sock->sk)
1493                 con->sock->sk->sk_user_data = NULL;
1494 }
1495
1496 static void free_conn(struct connection *con)
1497 {
1498         close_connection(con, true);
1499         if (con->othercon)
1500                 kmem_cache_free(con_cache, con->othercon);
1501         hlist_del(&con->list);
1502         kmem_cache_free(con_cache, con);
1503 }
1504
1505 void dlm_lowcomms_stop(void)
1506 {
1507         /* Set all the flags to prevent any
1508            socket activity.
1509         */
1510         mutex_lock(&connections_lock);
1511         foreach_conn(stop_conn);
1512         mutex_unlock(&connections_lock);
1513
1514         work_stop();
1515
1516         mutex_lock(&connections_lock);
1517         clean_writequeues();
1518
1519         foreach_conn(free_conn);
1520
1521         mutex_unlock(&connections_lock);
1522         kmem_cache_destroy(con_cache);
1523 }
1524
1525 int dlm_lowcomms_start(void)
1526 {
1527         int error = -EINVAL;
1528         struct connection *con;
1529         int i;
1530
1531         for (i = 0; i < CONN_HASH_SIZE; i++)
1532                 INIT_HLIST_HEAD(&connection_hash[i]);
1533
1534         init_local();
1535         if (!dlm_local_count) {
1536                 error = -ENOTCONN;
1537                 log_print("no local IP address has been set");
1538                 goto out;
1539         }
1540
1541         error = -ENOMEM;
1542         con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
1543                                       __alignof__(struct connection), 0,
1544                                       NULL);
1545         if (!con_cache)
1546                 goto out;
1547
1548         /* Start listening */
1549         if (dlm_config.ci_protocol == 0)
1550                 error = tcp_listen_for_all();
1551         else
1552                 error = sctp_listen_for_all();
1553         if (error)
1554                 goto fail_unlisten;
1555
1556         error = work_start();
1557         if (error)
1558                 goto fail_unlisten;
1559
1560         return 0;
1561
1562 fail_unlisten:
1563         con = nodeid2con(0,0);
1564         if (con) {
1565                 close_connection(con, false);
1566                 kmem_cache_free(con_cache, con);
1567         }
1568         kmem_cache_destroy(con_cache);
1569
1570 out:
1571         return error;
1572 }