afs: Remote abort can cause BUG in rxrpc code
[linux-2.6.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                 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                                 unsigned char *b=(unsigned char *)&prim.ssp_addr;
516                                 log_print("reject connect from unknown addr");
517                                 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 
518                                                      b, sizeof(struct sockaddr_storage));
519                                 sctp_send_shutdown(prim.ssp_assoc_id);
520                                 return;
521                         }
522
523                         new_con = nodeid2con(nodeid, GFP_NOFS);
524                         if (!new_con)
525                                 return;
526
527                         /* Peel off a new sock */
528                         parg.associd = sn->sn_assoc_change.sac_assoc_id;
529                         ret = kernel_getsockopt(con->sock, IPPROTO_SCTP,
530                                                 SCTP_SOCKOPT_PEELOFF,
531                                                 (void *)&parg, &parglen);
532                         if (ret < 0) {
533                                 log_print("Can't peel off a socket for "
534                                           "connection %d to node %d: err=%d",
535                                           parg.associd, nodeid, ret);
536                                 return;
537                         }
538                         new_con->sock = sockfd_lookup(parg.sd, &err);
539                         if (!new_con->sock) {
540                                 log_print("sockfd_lookup error %d", err);
541                                 return;
542                         }
543                         add_sock(new_con->sock, new_con);
544                         sockfd_put(new_con->sock);
545
546                         log_print("connecting to %d sctp association %d",
547                                  nodeid, (int)sn->sn_assoc_change.sac_assoc_id);
548
549                         /* Send any pending writes */
550                         clear_bit(CF_CONNECT_PENDING, &new_con->flags);
551                         clear_bit(CF_INIT_PENDING, &con->flags);
552                         if (!test_and_set_bit(CF_WRITE_PENDING, &new_con->flags)) {
553                                 queue_work(send_workqueue, &new_con->swork);
554                         }
555                         if (!test_and_set_bit(CF_READ_PENDING, &new_con->flags))
556                                 queue_work(recv_workqueue, &new_con->rwork);
557                 }
558                 break;
559
560                 case SCTP_COMM_LOST:
561                 case SCTP_SHUTDOWN_COMP:
562                 {
563                         con = assoc2con(sn->sn_assoc_change.sac_assoc_id);
564                         if (con) {
565                                 con->sctp_assoc = 0;
566                         }
567                 }
568                 break;
569
570                 /* We don't know which INIT failed, so clear the PENDING flags
571                  * on them all.  if assoc_id is zero then it will then try
572                  * again */
573
574                 case SCTP_CANT_STR_ASSOC:
575                 {
576                         log_print("Can't start SCTP association - retrying");
577                         sctp_init_failed();
578                 }
579                 break;
580
581                 default:
582                         log_print("unexpected SCTP assoc change id=%d state=%d",
583                                   (int)sn->sn_assoc_change.sac_assoc_id,
584                                   sn->sn_assoc_change.sac_state);
585                 }
586         }
587 }
588
589 /* Data received from remote end */
590 static int receive_from_sock(struct connection *con)
591 {
592         int ret = 0;
593         struct msghdr msg = {};
594         struct kvec iov[2];
595         unsigned len;
596         int r;
597         int call_again_soon = 0;
598         int nvec;
599         char incmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
600
601         mutex_lock(&con->sock_mutex);
602
603         if (con->sock == NULL) {
604                 ret = -EAGAIN;
605                 goto out_close;
606         }
607
608         if (con->rx_page == NULL) {
609                 /*
610                  * This doesn't need to be atomic, but I think it should
611                  * improve performance if it is.
612                  */
613                 con->rx_page = alloc_page(GFP_ATOMIC);
614                 if (con->rx_page == NULL)
615                         goto out_resched;
616                 cbuf_init(&con->cb, PAGE_CACHE_SIZE);
617         }
618
619         /* Only SCTP needs these really */
620         memset(&incmsg, 0, sizeof(incmsg));
621         msg.msg_control = incmsg;
622         msg.msg_controllen = sizeof(incmsg);
623
624         /*
625          * iov[0] is the bit of the circular buffer between the current end
626          * point (cb.base + cb.len) and the end of the buffer.
627          */
628         iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
629         iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
630         iov[1].iov_len = 0;
631         nvec = 1;
632
633         /*
634          * iov[1] is the bit of the circular buffer between the start of the
635          * buffer and the start of the currently used section (cb.base)
636          */
637         if (cbuf_data(&con->cb) >= con->cb.base) {
638                 iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);
639                 iov[1].iov_len = con->cb.base;
640                 iov[1].iov_base = page_address(con->rx_page);
641                 nvec = 2;
642         }
643         len = iov[0].iov_len + iov[1].iov_len;
644
645         r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,
646                                MSG_DONTWAIT | MSG_NOSIGNAL);
647         if (ret <= 0)
648                 goto out_close;
649
650         /* Process SCTP notifications */
651         if (msg.msg_flags & MSG_NOTIFICATION) {
652                 msg.msg_control = incmsg;
653                 msg.msg_controllen = sizeof(incmsg);
654
655                 process_sctp_notification(con, &msg,
656                                 page_address(con->rx_page) + con->cb.base);
657                 mutex_unlock(&con->sock_mutex);
658                 return 0;
659         }
660         BUG_ON(con->nodeid == 0);
661
662         if (ret == len)
663                 call_again_soon = 1;
664         cbuf_add(&con->cb, ret);
665         ret = dlm_process_incoming_buffer(con->nodeid,
666                                           page_address(con->rx_page),
667                                           con->cb.base, con->cb.len,
668                                           PAGE_CACHE_SIZE);
669         if (ret == -EBADMSG) {
670                 log_print("lowcomms: addr=%p, base=%u, len=%u, "
671                           "iov_len=%u, iov_base[0]=%p, read=%d",
672                           page_address(con->rx_page), con->cb.base, con->cb.len,
673                           len, iov[0].iov_base, r);
674         }
675         if (ret < 0)
676                 goto out_close;
677         cbuf_eat(&con->cb, ret);
678
679         if (cbuf_empty(&con->cb) && !call_again_soon) {
680                 __free_page(con->rx_page);
681                 con->rx_page = NULL;
682         }
683
684         if (call_again_soon)
685                 goto out_resched;
686         mutex_unlock(&con->sock_mutex);
687         return 0;
688
689 out_resched:
690         if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
691                 queue_work(recv_workqueue, &con->rwork);
692         mutex_unlock(&con->sock_mutex);
693         return -EAGAIN;
694
695 out_close:
696         mutex_unlock(&con->sock_mutex);
697         if (ret != -EAGAIN) {
698                 close_connection(con, false);
699                 /* Reconnect when there is something to send */
700         }
701         /* Don't return success if we really got EOF */
702         if (ret == 0)
703                 ret = -EAGAIN;
704
705         return ret;
706 }
707
708 /* Listening socket is busy, accept a connection */
709 static int tcp_accept_from_sock(struct connection *con)
710 {
711         int result;
712         struct sockaddr_storage peeraddr;
713         struct socket *newsock;
714         int len;
715         int nodeid;
716         struct connection *newcon;
717         struct connection *addcon;
718
719         memset(&peeraddr, 0, sizeof(peeraddr));
720         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
721                                   IPPROTO_TCP, &newsock);
722         if (result < 0)
723                 return -ENOMEM;
724
725         mutex_lock_nested(&con->sock_mutex, 0);
726
727         result = -ENOTCONN;
728         if (con->sock == NULL)
729                 goto accept_err;
730
731         newsock->type = con->sock->type;
732         newsock->ops = con->sock->ops;
733
734         result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK);
735         if (result < 0)
736                 goto accept_err;
737
738         /* Get the connected socket's peer */
739         memset(&peeraddr, 0, sizeof(peeraddr));
740         if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr,
741                                   &len, 2)) {
742                 result = -ECONNABORTED;
743                 goto accept_err;
744         }
745
746         /* Get the new node's NODEID */
747         make_sockaddr(&peeraddr, 0, &len);
748         if (dlm_addr_to_nodeid(&peeraddr, &nodeid)) {
749                 unsigned char *b=(unsigned char *)&peeraddr;
750                 log_print("connect from non cluster node");
751                 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 
752                                      b, sizeof(struct sockaddr_storage));
753                 sock_release(newsock);
754                 mutex_unlock(&con->sock_mutex);
755                 return -1;
756         }
757
758         log_print("got connection from %d", nodeid);
759
760         /*  Check to see if we already have a connection to this node. This
761          *  could happen if the two nodes initiate a connection at roughly
762          *  the same time and the connections cross on the wire.
763          *  In this case we store the incoming one in "othercon"
764          */
765         newcon = nodeid2con(nodeid, GFP_NOFS);
766         if (!newcon) {
767                 result = -ENOMEM;
768                 goto accept_err;
769         }
770         mutex_lock_nested(&newcon->sock_mutex, 1);
771         if (newcon->sock) {
772                 struct connection *othercon = newcon->othercon;
773
774                 if (!othercon) {
775                         othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
776                         if (!othercon) {
777                                 log_print("failed to allocate incoming socket");
778                                 mutex_unlock(&newcon->sock_mutex);
779                                 result = -ENOMEM;
780                                 goto accept_err;
781                         }
782                         othercon->nodeid = nodeid;
783                         othercon->rx_action = receive_from_sock;
784                         mutex_init(&othercon->sock_mutex);
785                         INIT_WORK(&othercon->swork, process_send_sockets);
786                         INIT_WORK(&othercon->rwork, process_recv_sockets);
787                         set_bit(CF_IS_OTHERCON, &othercon->flags);
788                 }
789                 if (!othercon->sock) {
790                         newcon->othercon = othercon;
791                         othercon->sock = newsock;
792                         newsock->sk->sk_user_data = othercon;
793                         add_sock(newsock, othercon);
794                         addcon = othercon;
795                 }
796                 else {
797                         printk("Extra connection from node %d attempted\n", nodeid);
798                         result = -EAGAIN;
799                         mutex_unlock(&newcon->sock_mutex);
800                         goto accept_err;
801                 }
802         }
803         else {
804                 newsock->sk->sk_user_data = newcon;
805                 newcon->rx_action = receive_from_sock;
806                 add_sock(newsock, newcon);
807                 addcon = newcon;
808         }
809
810         mutex_unlock(&newcon->sock_mutex);
811
812         /*
813          * Add it to the active queue in case we got data
814          * between processing the accept adding the socket
815          * to the read_sockets list
816          */
817         if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
818                 queue_work(recv_workqueue, &addcon->rwork);
819         mutex_unlock(&con->sock_mutex);
820
821         return 0;
822
823 accept_err:
824         mutex_unlock(&con->sock_mutex);
825         sock_release(newsock);
826
827         if (result != -EAGAIN)
828                 log_print("error accepting connection from node: %d", result);
829         return result;
830 }
831
832 static void free_entry(struct writequeue_entry *e)
833 {
834         __free_page(e->page);
835         kfree(e);
836 }
837
838 /* Initiate an SCTP association.
839    This is a special case of send_to_sock() in that we don't yet have a
840    peeled-off socket for this association, so we use the listening socket
841    and add the primary IP address of the remote node.
842  */
843 static void sctp_init_assoc(struct connection *con)
844 {
845         struct sockaddr_storage rem_addr;
846         char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
847         struct msghdr outmessage;
848         struct cmsghdr *cmsg;
849         struct sctp_sndrcvinfo *sinfo;
850         struct connection *base_con;
851         struct writequeue_entry *e;
852         int len, offset;
853         int ret;
854         int addrlen;
855         struct kvec iov[1];
856
857         if (test_and_set_bit(CF_INIT_PENDING, &con->flags))
858                 return;
859
860         if (con->retries++ > MAX_CONNECT_RETRIES)
861                 return;
862
863         if (nodeid_to_addr(con->nodeid, (struct sockaddr *)&rem_addr)) {
864                 log_print("no address for nodeid %d", con->nodeid);
865                 return;
866         }
867         base_con = nodeid2con(0, 0);
868         BUG_ON(base_con == NULL);
869
870         make_sockaddr(&rem_addr, dlm_config.ci_tcp_port, &addrlen);
871
872         outmessage.msg_name = &rem_addr;
873         outmessage.msg_namelen = addrlen;
874         outmessage.msg_control = outcmsg;
875         outmessage.msg_controllen = sizeof(outcmsg);
876         outmessage.msg_flags = MSG_EOR;
877
878         spin_lock(&con->writequeue_lock);
879
880         if (list_empty(&con->writequeue)) {
881                 spin_unlock(&con->writequeue_lock);
882                 log_print("writequeue empty for nodeid %d", con->nodeid);
883                 return;
884         }
885
886         e = list_first_entry(&con->writequeue, struct writequeue_entry, list);
887         len = e->len;
888         offset = e->offset;
889         spin_unlock(&con->writequeue_lock);
890
891         /* Send the first block off the write queue */
892         iov[0].iov_base = page_address(e->page)+offset;
893         iov[0].iov_len = len;
894
895         cmsg = CMSG_FIRSTHDR(&outmessage);
896         cmsg->cmsg_level = IPPROTO_SCTP;
897         cmsg->cmsg_type = SCTP_SNDRCV;
898         cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
899         sinfo = CMSG_DATA(cmsg);
900         memset(sinfo, 0x00, sizeof(struct sctp_sndrcvinfo));
901         sinfo->sinfo_ppid = cpu_to_le32(dlm_our_nodeid());
902         outmessage.msg_controllen = cmsg->cmsg_len;
903
904         ret = kernel_sendmsg(base_con->sock, &outmessage, iov, 1, len);
905         if (ret < 0) {
906                 log_print("Send first packet to node %d failed: %d",
907                           con->nodeid, ret);
908
909                 /* Try again later */
910                 clear_bit(CF_CONNECT_PENDING, &con->flags);
911                 clear_bit(CF_INIT_PENDING, &con->flags);
912         }
913         else {
914                 spin_lock(&con->writequeue_lock);
915                 e->offset += ret;
916                 e->len -= ret;
917
918                 if (e->len == 0 && e->users == 0) {
919                         list_del(&e->list);
920                         free_entry(e);
921                 }
922                 spin_unlock(&con->writequeue_lock);
923         }
924 }
925
926 /* Connect a new socket to its peer */
927 static void tcp_connect_to_sock(struct connection *con)
928 {
929         int result = -EHOSTUNREACH;
930         struct sockaddr_storage saddr, src_addr;
931         int addr_len;
932         struct socket *sock = NULL;
933         int one = 1;
934
935         if (con->nodeid == 0) {
936                 log_print("attempt to connect sock 0 foiled");
937                 return;
938         }
939
940         mutex_lock(&con->sock_mutex);
941         if (con->retries++ > MAX_CONNECT_RETRIES)
942                 goto out;
943
944         /* Some odd races can cause double-connects, ignore them */
945         if (con->sock) {
946                 result = 0;
947                 goto out;
948         }
949
950         /* Create a socket to communicate with */
951         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
952                                   IPPROTO_TCP, &sock);
953         if (result < 0)
954                 goto out_err;
955
956         memset(&saddr, 0, sizeof(saddr));
957         if (dlm_nodeid_to_addr(con->nodeid, &saddr))
958                 goto out_err;
959
960         sock->sk->sk_user_data = con;
961         con->rx_action = receive_from_sock;
962         con->connect_action = tcp_connect_to_sock;
963         add_sock(sock, con);
964
965         /* Bind to our cluster-known address connecting to avoid
966            routing problems */
967         memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
968         make_sockaddr(&src_addr, 0, &addr_len);
969         result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
970                                  addr_len);
971         if (result < 0) {
972                 log_print("could not bind for connect: %d", result);
973                 /* This *may* not indicate a critical error */
974         }
975
976         make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
977
978         log_print("connecting to %d", con->nodeid);
979
980         /* Turn off Nagle's algorithm */
981         kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
982                           sizeof(one));
983
984         result =
985                 sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
986                                    O_NONBLOCK);
987         if (result == -EINPROGRESS)
988                 result = 0;
989         if (result == 0)
990                 goto out;
991
992 out_err:
993         if (con->sock) {
994                 sock_release(con->sock);
995                 con->sock = NULL;
996         } else if (sock) {
997                 sock_release(sock);
998         }
999         /*
1000          * Some errors are fatal and this list might need adjusting. For other
1001          * errors we try again until the max number of retries is reached.
1002          */
1003         if (result != -EHOSTUNREACH && result != -ENETUNREACH &&
1004             result != -ENETDOWN && result != -EINVAL
1005             && result != -EPROTONOSUPPORT) {
1006                 lowcomms_connect_sock(con);
1007                 result = 0;
1008         }
1009 out:
1010         mutex_unlock(&con->sock_mutex);
1011         return;
1012 }
1013
1014 static struct socket *tcp_create_listen_sock(struct connection *con,
1015                                              struct sockaddr_storage *saddr)
1016 {
1017         struct socket *sock = NULL;
1018         int result = 0;
1019         int one = 1;
1020         int addr_len;
1021
1022         if (dlm_local_addr[0]->ss_family == AF_INET)
1023                 addr_len = sizeof(struct sockaddr_in);
1024         else
1025                 addr_len = sizeof(struct sockaddr_in6);
1026
1027         /* Create a socket to communicate with */
1028         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
1029                                   IPPROTO_TCP, &sock);
1030         if (result < 0) {
1031                 log_print("Can't create listening comms socket");
1032                 goto create_out;
1033         }
1034
1035         /* Turn off Nagle's algorithm */
1036         kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1037                           sizeof(one));
1038
1039         result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
1040                                    (char *)&one, sizeof(one));
1041
1042         if (result < 0) {
1043                 log_print("Failed to set SO_REUSEADDR on socket: %d", result);
1044         }
1045         sock->sk->sk_user_data = con;
1046         con->rx_action = tcp_accept_from_sock;
1047         con->connect_action = tcp_connect_to_sock;
1048         con->sock = sock;
1049
1050         /* Bind to our port */
1051         make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
1052         result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
1053         if (result < 0) {
1054                 log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
1055                 sock_release(sock);
1056                 sock = NULL;
1057                 con->sock = NULL;
1058                 goto create_out;
1059         }
1060         result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
1061                                  (char *)&one, sizeof(one));
1062         if (result < 0) {
1063                 log_print("Set keepalive failed: %d", result);
1064         }
1065
1066         result = sock->ops->listen(sock, 5);
1067         if (result < 0) {
1068                 log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
1069                 sock_release(sock);
1070                 sock = NULL;
1071                 goto create_out;
1072         }
1073
1074 create_out:
1075         return sock;
1076 }
1077
1078 /* Get local addresses */
1079 static void init_local(void)
1080 {
1081         struct sockaddr_storage sas, *addr;
1082         int i;
1083
1084         dlm_local_count = 0;
1085         for (i = 0; i < DLM_MAX_ADDR_COUNT - 1; i++) {
1086                 if (dlm_our_addr(&sas, i))
1087                         break;
1088
1089                 addr = kmalloc(sizeof(*addr), GFP_NOFS);
1090                 if (!addr)
1091                         break;
1092                 memcpy(addr, &sas, sizeof(*addr));
1093                 dlm_local_addr[dlm_local_count++] = addr;
1094         }
1095 }
1096
1097 /* Bind to an IP address. SCTP allows multiple address so it can do
1098    multi-homing */
1099 static int add_sctp_bind_addr(struct connection *sctp_con,
1100                               struct sockaddr_storage *addr,
1101                               int addr_len, int num)
1102 {
1103         int result = 0;
1104
1105         if (num == 1)
1106                 result = kernel_bind(sctp_con->sock,
1107                                      (struct sockaddr *) addr,
1108                                      addr_len);
1109         else
1110                 result = kernel_setsockopt(sctp_con->sock, SOL_SCTP,
1111                                            SCTP_SOCKOPT_BINDX_ADD,
1112                                            (char *)addr, addr_len);
1113
1114         if (result < 0)
1115                 log_print("Can't bind to port %d addr number %d",
1116                           dlm_config.ci_tcp_port, num);
1117
1118         return result;
1119 }
1120
1121 /* Initialise SCTP socket and bind to all interfaces */
1122 static int sctp_listen_for_all(void)
1123 {
1124         struct socket *sock = NULL;
1125         struct sockaddr_storage localaddr;
1126         struct sctp_event_subscribe subscribe;
1127         int result = -EINVAL, num = 1, i, addr_len;
1128         struct connection *con = nodeid2con(0, GFP_NOFS);
1129         int bufsize = NEEDED_RMEM;
1130
1131         if (!con)
1132                 return -ENOMEM;
1133
1134         log_print("Using SCTP for communications");
1135
1136         result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_SEQPACKET,
1137                                   IPPROTO_SCTP, &sock);
1138         if (result < 0) {
1139                 log_print("Can't create comms socket, check SCTP is loaded");
1140                 goto out;
1141         }
1142
1143         /* Listen for events */
1144         memset(&subscribe, 0, sizeof(subscribe));
1145         subscribe.sctp_data_io_event = 1;
1146         subscribe.sctp_association_event = 1;
1147         subscribe.sctp_send_failure_event = 1;
1148         subscribe.sctp_shutdown_event = 1;
1149         subscribe.sctp_partial_delivery_event = 1;
1150
1151         result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
1152                                  (char *)&bufsize, sizeof(bufsize));
1153         if (result)
1154                 log_print("Error increasing buffer space on socket %d", result);
1155
1156         result = kernel_setsockopt(sock, SOL_SCTP, SCTP_EVENTS,
1157                                    (char *)&subscribe, sizeof(subscribe));
1158         if (result < 0) {
1159                 log_print("Failed to set SCTP_EVENTS on socket: result=%d",
1160                           result);
1161                 goto create_delsock;
1162         }
1163
1164         /* Init con struct */
1165         sock->sk->sk_user_data = con;
1166         con->sock = sock;
1167         con->sock->sk->sk_data_ready = lowcomms_data_ready;
1168         con->rx_action = receive_from_sock;
1169         con->connect_action = sctp_init_assoc;
1170
1171         /* Bind to all interfaces. */
1172         for (i = 0; i < dlm_local_count; i++) {
1173                 memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr));
1174                 make_sockaddr(&localaddr, dlm_config.ci_tcp_port, &addr_len);
1175
1176                 result = add_sctp_bind_addr(con, &localaddr, addr_len, num);
1177                 if (result)
1178                         goto create_delsock;
1179                 ++num;
1180         }
1181
1182         result = sock->ops->listen(sock, 5);
1183         if (result < 0) {
1184                 log_print("Can't set socket listening");
1185                 goto create_delsock;
1186         }
1187
1188         return 0;
1189
1190 create_delsock:
1191         sock_release(sock);
1192         con->sock = NULL;
1193 out:
1194         return result;
1195 }
1196
1197 static int tcp_listen_for_all(void)
1198 {
1199         struct socket *sock = NULL;
1200         struct connection *con = nodeid2con(0, GFP_NOFS);
1201         int result = -EINVAL;
1202
1203         if (!con)
1204                 return -ENOMEM;
1205
1206         /* We don't support multi-homed hosts */
1207         if (dlm_local_addr[1] != NULL) {
1208                 log_print("TCP protocol can't handle multi-homed hosts, "
1209                           "try SCTP");
1210                 return -EINVAL;
1211         }
1212
1213         log_print("Using TCP for communications");
1214
1215         sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
1216         if (sock) {
1217                 add_sock(sock, con);
1218                 result = 0;
1219         }
1220         else {
1221                 result = -EADDRINUSE;
1222         }
1223
1224         return result;
1225 }
1226
1227
1228
1229 static struct writequeue_entry *new_writequeue_entry(struct connection *con,
1230                                                      gfp_t allocation)
1231 {
1232         struct writequeue_entry *entry;
1233
1234         entry = kmalloc(sizeof(struct writequeue_entry), allocation);
1235         if (!entry)
1236                 return NULL;
1237
1238         entry->page = alloc_page(allocation);
1239         if (!entry->page) {
1240                 kfree(entry);
1241                 return NULL;
1242         }
1243
1244         entry->offset = 0;
1245         entry->len = 0;
1246         entry->end = 0;
1247         entry->users = 0;
1248         entry->con = con;
1249
1250         return entry;
1251 }
1252
1253 void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
1254 {
1255         struct connection *con;
1256         struct writequeue_entry *e;
1257         int offset = 0;
1258         int users = 0;
1259
1260         con = nodeid2con(nodeid, allocation);
1261         if (!con)
1262                 return NULL;
1263
1264         spin_lock(&con->writequeue_lock);
1265         e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
1266         if ((&e->list == &con->writequeue) ||
1267             (PAGE_CACHE_SIZE - e->end < len)) {
1268                 e = NULL;
1269         } else {
1270                 offset = e->end;
1271                 e->end += len;
1272                 users = e->users++;
1273         }
1274         spin_unlock(&con->writequeue_lock);
1275
1276         if (e) {
1277         got_one:
1278                 *ppc = page_address(e->page) + offset;
1279                 return e;
1280         }
1281
1282         e = new_writequeue_entry(con, allocation);
1283         if (e) {
1284                 spin_lock(&con->writequeue_lock);
1285                 offset = e->end;
1286                 e->end += len;
1287                 users = e->users++;
1288                 list_add_tail(&e->list, &con->writequeue);
1289                 spin_unlock(&con->writequeue_lock);
1290                 goto got_one;
1291         }
1292         return NULL;
1293 }
1294
1295 void dlm_lowcomms_commit_buffer(void *mh)
1296 {
1297         struct writequeue_entry *e = (struct writequeue_entry *)mh;
1298         struct connection *con = e->con;
1299         int users;
1300
1301         spin_lock(&con->writequeue_lock);
1302         users = --e->users;
1303         if (users)
1304                 goto out;
1305         e->len = e->end - e->offset;
1306         spin_unlock(&con->writequeue_lock);
1307
1308         if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {
1309                 queue_work(send_workqueue, &con->swork);
1310         }
1311         return;
1312
1313 out:
1314         spin_unlock(&con->writequeue_lock);
1315         return;
1316 }
1317
1318 /* Send a message */
1319 static void send_to_sock(struct connection *con)
1320 {
1321         int ret = 0;
1322         const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
1323         struct writequeue_entry *e;
1324         int len, offset;
1325         int count = 0;
1326
1327         mutex_lock(&con->sock_mutex);
1328         if (con->sock == NULL)
1329                 goto out_connect;
1330
1331         spin_lock(&con->writequeue_lock);
1332         for (;;) {
1333                 e = list_entry(con->writequeue.next, struct writequeue_entry,
1334                                list);
1335                 if ((struct list_head *) e == &con->writequeue)
1336                         break;
1337
1338                 len = e->len;
1339                 offset = e->offset;
1340                 BUG_ON(len == 0 && e->users == 0);
1341                 spin_unlock(&con->writequeue_lock);
1342
1343                 ret = 0;
1344                 if (len) {
1345                         ret = kernel_sendpage(con->sock, e->page, offset, len,
1346                                               msg_flags);
1347                         if (ret == -EAGAIN || ret == 0) {
1348                                 if (ret == -EAGAIN &&
1349                                     test_bit(SOCK_ASYNC_NOSPACE, &con->sock->flags) &&
1350                                     !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1351                                         /* Notify TCP that we're limited by the
1352                                          * application window size.
1353                                          */
1354                                         set_bit(SOCK_NOSPACE, &con->sock->flags);
1355                                         con->sock->sk->sk_write_pending++;
1356                                 }
1357                                 cond_resched();
1358                                 goto out;
1359                         }
1360                         if (ret <= 0)
1361                                 goto send_error;
1362                 }
1363
1364                 /* Don't starve people filling buffers */
1365                 if (++count >= MAX_SEND_MSG_COUNT) {
1366                         cond_resched();
1367                         count = 0;
1368                 }
1369
1370                 spin_lock(&con->writequeue_lock);
1371                 e->offset += ret;
1372                 e->len -= ret;
1373
1374                 if (e->len == 0 && e->users == 0) {
1375                         list_del(&e->list);
1376                         free_entry(e);
1377                         continue;
1378                 }
1379         }
1380         spin_unlock(&con->writequeue_lock);
1381 out:
1382         mutex_unlock(&con->sock_mutex);
1383         return;
1384
1385 send_error:
1386         mutex_unlock(&con->sock_mutex);
1387         close_connection(con, false);
1388         lowcomms_connect_sock(con);
1389         return;
1390
1391 out_connect:
1392         mutex_unlock(&con->sock_mutex);
1393         if (!test_bit(CF_INIT_PENDING, &con->flags))
1394                 lowcomms_connect_sock(con);
1395         return;
1396 }
1397
1398 static void clean_one_writequeue(struct connection *con)
1399 {
1400         struct writequeue_entry *e, *safe;
1401
1402         spin_lock(&con->writequeue_lock);
1403         list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1404                 list_del(&e->list);
1405                 free_entry(e);
1406         }
1407         spin_unlock(&con->writequeue_lock);
1408 }
1409
1410 /* Called from recovery when it knows that a node has
1411    left the cluster */
1412 int dlm_lowcomms_close(int nodeid)
1413 {
1414         struct connection *con;
1415
1416         log_print("closing connection to node %d", nodeid);
1417         con = nodeid2con(nodeid, 0);
1418         if (con) {
1419                 clear_bit(CF_CONNECT_PENDING, &con->flags);
1420                 clear_bit(CF_WRITE_PENDING, &con->flags);
1421                 set_bit(CF_CLOSE, &con->flags);
1422                 if (cancel_work_sync(&con->swork))
1423                         log_print("canceled swork for node %d", nodeid);
1424                 if (cancel_work_sync(&con->rwork))
1425                         log_print("canceled rwork for node %d", nodeid);
1426                 clean_one_writequeue(con);
1427                 close_connection(con, true);
1428         }
1429         return 0;
1430 }
1431
1432 /* Receive workqueue function */
1433 static void process_recv_sockets(struct work_struct *work)
1434 {
1435         struct connection *con = container_of(work, struct connection, rwork);
1436         int err;
1437
1438         clear_bit(CF_READ_PENDING, &con->flags);
1439         do {
1440                 err = con->rx_action(con);
1441         } while (!err);
1442 }
1443
1444 /* Send workqueue function */
1445 static void process_send_sockets(struct work_struct *work)
1446 {
1447         struct connection *con = container_of(work, struct connection, swork);
1448
1449         if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
1450                 con->connect_action(con);
1451                 set_bit(CF_WRITE_PENDING, &con->flags);
1452         }
1453         if (test_and_clear_bit(CF_WRITE_PENDING, &con->flags))
1454                 send_to_sock(con);
1455 }
1456
1457
1458 /* Discard all entries on the write queues */
1459 static void clean_writequeues(void)
1460 {
1461         foreach_conn(clean_one_writequeue);
1462 }
1463
1464 static void work_stop(void)
1465 {
1466         destroy_workqueue(recv_workqueue);
1467         destroy_workqueue(send_workqueue);
1468 }
1469
1470 static int work_start(void)
1471 {
1472         recv_workqueue = alloc_workqueue("dlm_recv",
1473                                          WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1474         if (!recv_workqueue) {
1475                 log_print("can't start dlm_recv");
1476                 return -ENOMEM;
1477         }
1478
1479         send_workqueue = alloc_workqueue("dlm_send",
1480                                          WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1481         if (!send_workqueue) {
1482                 log_print("can't start dlm_send");
1483                 destroy_workqueue(recv_workqueue);
1484                 return -ENOMEM;
1485         }
1486
1487         return 0;
1488 }
1489
1490 static void stop_conn(struct connection *con)
1491 {
1492         con->flags |= 0x0F;
1493         if (con->sock && con->sock->sk)
1494                 con->sock->sk->sk_user_data = NULL;
1495 }
1496
1497 static void free_conn(struct connection *con)
1498 {
1499         close_connection(con, true);
1500         if (con->othercon)
1501                 kmem_cache_free(con_cache, con->othercon);
1502         hlist_del(&con->list);
1503         kmem_cache_free(con_cache, con);
1504 }
1505
1506 void dlm_lowcomms_stop(void)
1507 {
1508         /* Set all the flags to prevent any
1509            socket activity.
1510         */
1511         mutex_lock(&connections_lock);
1512         foreach_conn(stop_conn);
1513         mutex_unlock(&connections_lock);
1514
1515         work_stop();
1516
1517         mutex_lock(&connections_lock);
1518         clean_writequeues();
1519
1520         foreach_conn(free_conn);
1521
1522         mutex_unlock(&connections_lock);
1523         kmem_cache_destroy(con_cache);
1524 }
1525
1526 int dlm_lowcomms_start(void)
1527 {
1528         int error = -EINVAL;
1529         struct connection *con;
1530         int i;
1531
1532         for (i = 0; i < CONN_HASH_SIZE; i++)
1533                 INIT_HLIST_HEAD(&connection_hash[i]);
1534
1535         init_local();
1536         if (!dlm_local_count) {
1537                 error = -ENOTCONN;
1538                 log_print("no local IP address has been set");
1539                 goto out;
1540         }
1541
1542         error = -ENOMEM;
1543         con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
1544                                       __alignof__(struct connection), 0,
1545                                       NULL);
1546         if (!con_cache)
1547                 goto out;
1548
1549         /* Start listening */
1550         if (dlm_config.ci_protocol == 0)
1551                 error = tcp_listen_for_all();
1552         else
1553                 error = sctp_listen_for_all();
1554         if (error)
1555                 goto fail_unlisten;
1556
1557         error = work_start();
1558         if (error)
1559                 goto fail_unlisten;
1560
1561         return 0;
1562
1563 fail_unlisten:
1564         con = nodeid2con(0,0);
1565         if (con) {
1566                 close_connection(con, false);
1567                 kmem_cache_free(con_cache, con);
1568         }
1569         kmem_cache_destroy(con_cache);
1570
1571 out:
1572         return error;
1573 }