aboutsummaryrefslogtreecommitdiffstats
path: root/fs/dlm/lowcomms-tcp.c
diff options
context:
space:
mode:
Diffstat (limited to 'fs/dlm/lowcomms-tcp.c')
-rw-r--r--fs/dlm/lowcomms-tcp.c1189
1 files changed, 1189 insertions, 0 deletions
diff --git a/fs/dlm/lowcomms-tcp.c b/fs/dlm/lowcomms-tcp.c
new file mode 100644
index 000000000000..8f2791fc8447
--- /dev/null
+++ b/fs/dlm/lowcomms-tcp.c
@@ -0,0 +1,1189 @@
1/******************************************************************************
2*******************************************************************************
3**
4** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5** Copyright (C) 2004-2006 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 it's
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 * I don't see any problem with the recv thread executing the locking
40 * code on behalf of remote processes as the locking code is
41 * short, efficient and never waits.
42 *
43 */
44
45
46#include <asm/ioctls.h>
47#include <net/sock.h>
48#include <net/tcp.h>
49#include <linux/pagemap.h>
50
51#include "dlm_internal.h"
52#include "lowcomms.h"
53#include "midcomms.h"
54#include "config.h"
55
56struct cbuf {
57 unsigned int base;
58 unsigned int len;
59 unsigned int mask;
60};
61
62#define NODE_INCREMENT 32
63static void cbuf_add(struct cbuf *cb, int n)
64{
65 cb->len += n;
66}
67
68static int cbuf_data(struct cbuf *cb)
69{
70 return ((cb->base + cb->len) & cb->mask);
71}
72
73static void cbuf_init(struct cbuf *cb, int size)
74{
75 cb->base = cb->len = 0;
76 cb->mask = size-1;
77}
78
79static void cbuf_eat(struct cbuf *cb, int n)
80{
81 cb->len -= n;
82 cb->base += n;
83 cb->base &= cb->mask;
84}
85
86static bool cbuf_empty(struct cbuf *cb)
87{
88 return cb->len == 0;
89}
90
91/* Maximum number of incoming messages to process before
92 doing a cond_resched()
93*/
94#define MAX_RX_MSG_COUNT 25
95
96struct connection {
97 struct socket *sock; /* NULL if not connected */
98 uint32_t nodeid; /* So we know who we are in the list */
99 struct rw_semaphore sock_sem; /* Stop connect races */
100 struct list_head read_list; /* On this list when ready for reading */
101 struct list_head write_list; /* On this list when ready for writing */
102 struct list_head state_list; /* On this list when ready to connect */
103 unsigned long flags; /* bit 1,2 = We are on the read/write lists */
104#define CF_READ_PENDING 1
105#define CF_WRITE_PENDING 2
106#define CF_CONNECT_PENDING 3
107#define CF_IS_OTHERCON 4
108 struct list_head writequeue; /* List of outgoing writequeue_entries */
109 struct list_head listenlist; /* List of allocated listening sockets */
110 spinlock_t writequeue_lock;
111 int (*rx_action) (struct connection *); /* What to do when active */
112 struct page *rx_page;
113 struct cbuf cb;
114 int retries;
115 atomic_t waiting_requests;
116#define MAX_CONNECT_RETRIES 3
117 struct connection *othercon;
118};
119#define sock2con(x) ((struct connection *)(x)->sk_user_data)
120
121/* An entry waiting to be sent */
122struct writequeue_entry {
123 struct list_head list;
124 struct page *page;
125 int offset;
126 int len;
127 int end;
128 int users;
129 struct connection *con;
130};
131
132static struct sockaddr_storage dlm_local_addr;
133
134/* Manage daemons */
135static struct task_struct *recv_task;
136static struct task_struct *send_task;
137
138static wait_queue_t lowcomms_send_waitq_head;
139static DECLARE_WAIT_QUEUE_HEAD(lowcomms_send_waitq);
140static wait_queue_t lowcomms_recv_waitq_head;
141static DECLARE_WAIT_QUEUE_HEAD(lowcomms_recv_waitq);
142
143/* An array of pointers to connections, indexed by NODEID */
144static struct connection **connections;
145static DECLARE_MUTEX(connections_lock);
146static kmem_cache_t *con_cache;
147static int conn_array_size;
148
149/* List of sockets that have reads pending */
150static LIST_HEAD(read_sockets);
151static DEFINE_SPINLOCK(read_sockets_lock);
152
153/* List of sockets which have writes pending */
154static LIST_HEAD(write_sockets);
155static DEFINE_SPINLOCK(write_sockets_lock);
156
157/* List of sockets which have connects pending */
158static LIST_HEAD(state_sockets);
159static DEFINE_SPINLOCK(state_sockets_lock);
160
161static struct connection *nodeid2con(int nodeid, gfp_t allocation)
162{
163 struct connection *con = NULL;
164
165 down(&connections_lock);
166 if (nodeid >= conn_array_size) {
167 int new_size = nodeid + NODE_INCREMENT;
168 struct connection **new_conns;
169
170 new_conns = kzalloc(sizeof(struct connection *) *
171 new_size, allocation);
172 if (!new_conns)
173 goto finish;
174
175 memcpy(new_conns, connections, sizeof(struct connection *) * conn_array_size);
176 conn_array_size = new_size;
177 kfree(connections);
178 connections = new_conns;
179
180 }
181
182 con = connections[nodeid];
183 if (con == NULL && allocation) {
184 con = kmem_cache_zalloc(con_cache, allocation);
185 if (!con)
186 goto finish;
187
188 con->nodeid = nodeid;
189 init_rwsem(&con->sock_sem);
190 INIT_LIST_HEAD(&con->writequeue);
191 spin_lock_init(&con->writequeue_lock);
192
193 connections[nodeid] = con;
194 }
195
196finish:
197 up(&connections_lock);
198 return con;
199}
200
201/* Data available on socket or listen socket received a connect */
202static void lowcomms_data_ready(struct sock *sk, int count_unused)
203{
204 struct connection *con = sock2con(sk);
205
206 atomic_inc(&con->waiting_requests);
207 if (test_and_set_bit(CF_READ_PENDING, &con->flags))
208 return;
209
210 spin_lock_bh(&read_sockets_lock);
211 list_add_tail(&con->read_list, &read_sockets);
212 spin_unlock_bh(&read_sockets_lock);
213
214 wake_up_interruptible(&lowcomms_recv_waitq);
215}
216
217static void lowcomms_write_space(struct sock *sk)
218{
219 struct connection *con = sock2con(sk);
220
221 if (test_and_set_bit(CF_WRITE_PENDING, &con->flags))
222 return;
223
224 spin_lock_bh(&write_sockets_lock);
225 list_add_tail(&con->write_list, &write_sockets);
226 spin_unlock_bh(&write_sockets_lock);
227
228 wake_up_interruptible(&lowcomms_send_waitq);
229}
230
231static inline void lowcomms_connect_sock(struct connection *con)
232{
233 if (test_and_set_bit(CF_CONNECT_PENDING, &con->flags))
234 return;
235
236 spin_lock_bh(&state_sockets_lock);
237 list_add_tail(&con->state_list, &state_sockets);
238 spin_unlock_bh(&state_sockets_lock);
239
240 wake_up_interruptible(&lowcomms_send_waitq);
241}
242
243static void lowcomms_state_change(struct sock *sk)
244{
245 if (sk->sk_state == TCP_ESTABLISHED)
246 lowcomms_write_space(sk);
247}
248
249/* Make a socket active */
250static int add_sock(struct socket *sock, struct connection *con)
251{
252 con->sock = sock;
253
254 /* Install a data_ready callback */
255 con->sock->sk->sk_data_ready = lowcomms_data_ready;
256 con->sock->sk->sk_write_space = lowcomms_write_space;
257 con->sock->sk->sk_state_change = lowcomms_state_change;
258
259 return 0;
260}
261
262/* Add the port number to an IP6 or 4 sockaddr and return the address
263 length */
264static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
265 int *addr_len)
266{
267 saddr->ss_family = dlm_local_addr.ss_family;
268 if (saddr->ss_family == AF_INET) {
269 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
270 in4_addr->sin_port = cpu_to_be16(port);
271 *addr_len = sizeof(struct sockaddr_in);
272 } else {
273 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
274 in6_addr->sin6_port = cpu_to_be16(port);
275 *addr_len = sizeof(struct sockaddr_in6);
276 }
277}
278
279/* Close a remote connection and tidy up */
280static void close_connection(struct connection *con, bool and_other)
281{
282 down_write(&con->sock_sem);
283
284 if (con->sock) {
285 sock_release(con->sock);
286 con->sock = NULL;
287 }
288 if (con->othercon && and_other) {
289 /* Will only re-enter once. */
290 close_connection(con->othercon, false);
291 }
292 if (con->rx_page) {
293 __free_page(con->rx_page);
294 con->rx_page = NULL;
295 }
296 con->retries = 0;
297 up_write(&con->sock_sem);
298}
299
300/* Data received from remote end */
301static int receive_from_sock(struct connection *con)
302{
303 int ret = 0;
304 struct msghdr msg;
305 struct iovec iov[2];
306 mm_segment_t fs;
307 unsigned len;
308 int r;
309 int call_again_soon = 0;
310
311 down_read(&con->sock_sem);
312
313 if (con->sock == NULL)
314 goto out;
315 if (con->rx_page == NULL) {
316 /*
317 * This doesn't need to be atomic, but I think it should
318 * improve performance if it is.
319 */
320 con->rx_page = alloc_page(GFP_ATOMIC);
321 if (con->rx_page == NULL)
322 goto out_resched;
323 cbuf_init(&con->cb, PAGE_CACHE_SIZE);
324 }
325
326 msg.msg_control = NULL;
327 msg.msg_controllen = 0;
328 msg.msg_iovlen = 1;
329 msg.msg_iov = iov;
330 msg.msg_name = NULL;
331 msg.msg_namelen = 0;
332 msg.msg_flags = 0;
333
334 /*
335 * iov[0] is the bit of the circular buffer between the current end
336 * point (cb.base + cb.len) and the end of the buffer.
337 */
338 iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
339 iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
340 iov[1].iov_len = 0;
341
342 /*
343 * iov[1] is the bit of the circular buffer between the start of the
344 * buffer and the start of the currently used section (cb.base)
345 */
346 if (cbuf_data(&con->cb) >= con->cb.base) {
347 iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);
348 iov[1].iov_len = con->cb.base;
349 iov[1].iov_base = page_address(con->rx_page);
350 msg.msg_iovlen = 2;
351 }
352 len = iov[0].iov_len + iov[1].iov_len;
353
354 fs = get_fs();
355 set_fs(get_ds());
356 r = ret = sock_recvmsg(con->sock, &msg, len,
357 MSG_DONTWAIT | MSG_NOSIGNAL);
358 set_fs(fs);
359
360 if (ret <= 0)
361 goto out_close;
362 if (ret == len)
363 call_again_soon = 1;
364 cbuf_add(&con->cb, ret);
365 ret = dlm_process_incoming_buffer(con->nodeid,
366 page_address(con->rx_page),
367 con->cb.base, con->cb.len,
368 PAGE_CACHE_SIZE);
369 if (ret == -EBADMSG) {
370 printk(KERN_INFO "dlm: lowcomms: addr=%p, base=%u, len=%u, "
371 "iov_len=%u, iov_base[0]=%p, read=%d\n",
372 page_address(con->rx_page), con->cb.base, con->cb.len,
373 len, iov[0].iov_base, r);
374 }
375 if (ret < 0)
376 goto out_close;
377 cbuf_eat(&con->cb, ret);
378
379 if (cbuf_empty(&con->cb) && !call_again_soon) {
380 __free_page(con->rx_page);
381 con->rx_page = NULL;
382 }
383
384out:
385 if (call_again_soon)
386 goto out_resched;
387 up_read(&con->sock_sem);
388 return 0;
389
390out_resched:
391 lowcomms_data_ready(con->sock->sk, 0);
392 up_read(&con->sock_sem);
393 cond_resched();
394 return 0;
395
396out_close:
397 up_read(&con->sock_sem);
398 if (ret != -EAGAIN && !test_bit(CF_IS_OTHERCON, &con->flags)) {
399 close_connection(con, false);
400 /* Reconnect when there is something to send */
401 }
402
403 return ret;
404}
405
406/* Listening socket is busy, accept a connection */
407static int accept_from_sock(struct connection *con)
408{
409 int result;
410 struct sockaddr_storage peeraddr;
411 struct socket *newsock;
412 int len;
413 int nodeid;
414 struct connection *newcon;
415
416 memset(&peeraddr, 0, sizeof(peeraddr));
417 result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM,
418 IPPROTO_TCP, &newsock);
419 if (result < 0)
420 return -ENOMEM;
421
422 down_read(&con->sock_sem);
423
424 result = -ENOTCONN;
425 if (con->sock == NULL)
426 goto accept_err;
427
428 newsock->type = con->sock->type;
429 newsock->ops = con->sock->ops;
430
431 result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK);
432 if (result < 0)
433 goto accept_err;
434
435 /* Get the connected socket's peer */
436 memset(&peeraddr, 0, sizeof(peeraddr));
437 if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr,
438 &len, 2)) {
439 result = -ECONNABORTED;
440 goto accept_err;
441 }
442
443 /* Get the new node's NODEID */
444 make_sockaddr(&peeraddr, 0, &len);
445 if (dlm_addr_to_nodeid(&peeraddr, &nodeid)) {
446 printk("dlm: connect from non cluster node\n");
447 sock_release(newsock);
448 up_read(&con->sock_sem);
449 return -1;
450 }
451
452 log_print("got connection from %d", nodeid);
453
454 /* Check to see if we already have a connection to this node. This
455 * could happen if the two nodes initiate a connection at roughly
456 * the same time and the connections cross on the wire.
457 * TEMPORARY FIX:
458 * In this case we store the incoming one in "othercon"
459 */
460 newcon = nodeid2con(nodeid, GFP_KERNEL);
461 if (!newcon) {
462 result = -ENOMEM;
463 goto accept_err;
464 }
465 down_write(&newcon->sock_sem);
466 if (newcon->sock) {
467 struct connection *othercon = newcon->othercon;
468
469 if (!othercon) {
470 othercon = kmem_cache_zalloc(con_cache, GFP_KERNEL);
471 if (!othercon) {
472 printk("dlm: failed to allocate incoming socket\n");
473 up_write(&newcon->sock_sem);
474 result = -ENOMEM;
475 goto accept_err;
476 }
477 othercon->nodeid = nodeid;
478 othercon->rx_action = receive_from_sock;
479 init_rwsem(&othercon->sock_sem);
480 set_bit(CF_IS_OTHERCON, &othercon->flags);
481 newcon->othercon = othercon;
482 }
483 othercon->sock = newsock;
484 newsock->sk->sk_user_data = othercon;
485 add_sock(newsock, othercon);
486 }
487 else {
488 newsock->sk->sk_user_data = newcon;
489 newcon->rx_action = receive_from_sock;
490 add_sock(newsock, newcon);
491
492 }
493
494 up_write(&newcon->sock_sem);
495
496 /*
497 * Add it to the active queue in case we got data
498 * beween processing the accept adding the socket
499 * to the read_sockets list
500 */
501 lowcomms_data_ready(newsock->sk, 0);
502 up_read(&con->sock_sem);
503
504 return 0;
505
506accept_err:
507 up_read(&con->sock_sem);
508 sock_release(newsock);
509
510 if (result != -EAGAIN)
511 printk("dlm: error accepting connection from node: %d\n", result);
512 return result;
513}
514
515/* Connect a new socket to its peer */
516static void connect_to_sock(struct connection *con)
517{
518 int result = -EHOSTUNREACH;
519 struct sockaddr_storage saddr;
520 int addr_len;
521 struct socket *sock;
522
523 if (con->nodeid == 0) {
524 log_print("attempt to connect sock 0 foiled");
525 return;
526 }
527
528 down_write(&con->sock_sem);
529 if (con->retries++ > MAX_CONNECT_RETRIES)
530 goto out;
531
532 /* Some odd races can cause double-connects, ignore them */
533 if (con->sock) {
534 result = 0;
535 goto out;
536 }
537
538 /* Create a socket to communicate with */
539 result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM,
540 IPPROTO_TCP, &sock);
541 if (result < 0)
542 goto out_err;
543
544 memset(&saddr, 0, sizeof(saddr));
545 if (dlm_nodeid_to_addr(con->nodeid, &saddr))
546 goto out_err;
547
548 sock->sk->sk_user_data = con;
549 con->rx_action = receive_from_sock;
550
551 make_sockaddr(&saddr, dlm_config.tcp_port, &addr_len);
552
553 add_sock(sock, con);
554
555 log_print("connecting to %d", con->nodeid);
556 result =
557 sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
558 O_NONBLOCK);
559 if (result == -EINPROGRESS)
560 result = 0;
561 if (result == 0)
562 goto out;
563
564out_err:
565 if (con->sock) {
566 sock_release(con->sock);
567 con->sock = NULL;
568 }
569 /*
570 * Some errors are fatal and this list might need adjusting. For other
571 * errors we try again until the max number of retries is reached.
572 */
573 if (result != -EHOSTUNREACH && result != -ENETUNREACH &&
574 result != -ENETDOWN && result != EINVAL
575 && result != -EPROTONOSUPPORT) {
576 lowcomms_connect_sock(con);
577 result = 0;
578 }
579out:
580 up_write(&con->sock_sem);
581 return;
582}
583
584static struct socket *create_listen_sock(struct connection *con,
585 struct sockaddr_storage *saddr)
586{
587 struct socket *sock = NULL;
588 mm_segment_t fs;
589 int result = 0;
590 int one = 1;
591 int addr_len;
592
593 if (dlm_local_addr.ss_family == AF_INET)
594 addr_len = sizeof(struct sockaddr_in);
595 else
596 addr_len = sizeof(struct sockaddr_in6);
597
598 /* Create a socket to communicate with */
599 result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM, IPPROTO_TCP, &sock);
600 if (result < 0) {
601 printk("dlm: Can't create listening comms socket\n");
602 goto create_out;
603 }
604
605 fs = get_fs();
606 set_fs(get_ds());
607 result = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
608 (char *)&one, sizeof(one));
609 set_fs(fs);
610 if (result < 0) {
611 printk("dlm: Failed to set SO_REUSEADDR on socket: result=%d\n",
612 result);
613 }
614 sock->sk->sk_user_data = con;
615 con->rx_action = accept_from_sock;
616 con->sock = sock;
617
618 /* Bind to our port */
619 make_sockaddr(saddr, dlm_config.tcp_port, &addr_len);
620 result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
621 if (result < 0) {
622 printk("dlm: Can't bind to port %d\n", dlm_config.tcp_port);
623 sock_release(sock);
624 sock = NULL;
625 con->sock = NULL;
626 goto create_out;
627 }
628
629 fs = get_fs();
630 set_fs(get_ds());
631
632 result = sock_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
633 (char *)&one, sizeof(one));
634 set_fs(fs);
635 if (result < 0) {
636 printk("dlm: Set keepalive failed: %d\n", result);
637 }
638
639 result = sock->ops->listen(sock, 5);
640 if (result < 0) {
641 printk("dlm: Can't listen on port %d\n", dlm_config.tcp_port);
642 sock_release(sock);
643 sock = NULL;
644 goto create_out;
645 }
646
647create_out:
648 return sock;
649}
650
651
652/* Listen on all interfaces */
653static int listen_for_all(void)
654{
655 struct socket *sock = NULL;
656 struct connection *con = nodeid2con(0, GFP_KERNEL);
657 int result = -EINVAL;
658
659 /* We don't support multi-homed hosts */
660 set_bit(CF_IS_OTHERCON, &con->flags);
661
662 sock = create_listen_sock(con, &dlm_local_addr);
663 if (sock) {
664 add_sock(sock, con);
665 result = 0;
666 }
667 else {
668 result = -EADDRINUSE;
669 }
670
671 return result;
672}
673
674
675
676static struct writequeue_entry *new_writequeue_entry(struct connection *con,
677 gfp_t allocation)
678{
679 struct writequeue_entry *entry;
680
681 entry = kmalloc(sizeof(struct writequeue_entry), allocation);
682 if (!entry)
683 return NULL;
684
685 entry->page = alloc_page(allocation);
686 if (!entry->page) {
687 kfree(entry);
688 return NULL;
689 }
690
691 entry->offset = 0;
692 entry->len = 0;
693 entry->end = 0;
694 entry->users = 0;
695 entry->con = con;
696
697 return entry;
698}
699
700void *dlm_lowcomms_get_buffer(int nodeid, int len,
701 gfp_t allocation, char **ppc)
702{
703 struct connection *con;
704 struct writequeue_entry *e;
705 int offset = 0;
706 int users = 0;
707
708 con = nodeid2con(nodeid, allocation);
709 if (!con)
710 return NULL;
711
712 e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
713 if ((&e->list == &con->writequeue) ||
714 (PAGE_CACHE_SIZE - e->end < len)) {
715 e = NULL;
716 } else {
717 offset = e->end;
718 e->end += len;
719 users = e->users++;
720 }
721 spin_unlock(&con->writequeue_lock);
722
723 if (e) {
724 got_one:
725 if (users == 0)
726 kmap(e->page);
727 *ppc = page_address(e->page) + offset;
728 return e;
729 }
730
731 e = new_writequeue_entry(con, allocation);
732 if (e) {
733 spin_lock(&con->writequeue_lock);
734 offset = e->end;
735 e->end += len;
736 users = e->users++;
737 list_add_tail(&e->list, &con->writequeue);
738 spin_unlock(&con->writequeue_lock);
739 goto got_one;
740 }
741 return NULL;
742}
743
744void dlm_lowcomms_commit_buffer(void *mh)
745{
746 struct writequeue_entry *e = (struct writequeue_entry *)mh;
747 struct connection *con = e->con;
748 int users;
749
750 users = --e->users;
751 if (users)
752 goto out;
753 e->len = e->end - e->offset;
754 kunmap(e->page);
755 spin_unlock(&con->writequeue_lock);
756
757 if (test_and_set_bit(CF_WRITE_PENDING, &con->flags) == 0) {
758 spin_lock_bh(&write_sockets_lock);
759 list_add_tail(&con->write_list, &write_sockets);
760 spin_unlock_bh(&write_sockets_lock);
761
762 wake_up_interruptible(&lowcomms_send_waitq);
763 }
764 return;
765
766out:
767 spin_unlock(&con->writequeue_lock);
768 return;
769}
770
771static void free_entry(struct writequeue_entry *e)
772{
773 __free_page(e->page);
774 kfree(e);
775}
776
777/* Send a message */
778static void send_to_sock(struct connection *con)
779{
780 int ret = 0;
781 ssize_t(*sendpage) (struct socket *, struct page *, int, size_t, int);
782 const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
783 struct writequeue_entry *e;
784 int len, offset;
785
786 down_read(&con->sock_sem);
787 if (con->sock == NULL)
788 goto out_connect;
789
790 sendpage = con->sock->ops->sendpage;
791
792 spin_lock(&con->writequeue_lock);
793 for (;;) {
794 e = list_entry(con->writequeue.next, struct writequeue_entry,
795 list);
796 if ((struct list_head *) e == &con->writequeue)
797 break;
798
799 len = e->len;
800 offset = e->offset;
801 BUG_ON(len == 0 && e->users == 0);
802 spin_unlock(&con->writequeue_lock);
803
804 ret = 0;
805 if (len) {
806 ret = sendpage(con->sock, e->page, offset, len,
807 msg_flags);
808 if (ret == -EAGAIN || ret == 0)
809 goto out;
810 if (ret <= 0)
811 goto send_error;
812 }
813 else {
814 /* Don't starve people filling buffers */
815 cond_resched();
816 }
817
818 spin_lock(&con->writequeue_lock);
819 e->offset += ret;
820 e->len -= ret;
821
822 if (e->len == 0 && e->users == 0) {
823 list_del(&e->list);
824 kunmap(e->page);
825 free_entry(e);
826 continue;
827 }
828 }
829 spin_unlock(&con->writequeue_lock);
830out:
831 up_read(&con->sock_sem);
832 return;
833
834send_error:
835 up_read(&con->sock_sem);
836 close_connection(con, false);
837 lowcomms_connect_sock(con);
838 return;
839
840out_connect:
841 up_read(&con->sock_sem);
842 lowcomms_connect_sock(con);
843 return;
844}
845
846static void clean_one_writequeue(struct connection *con)
847{
848 struct list_head *list;
849 struct list_head *temp;
850
851 spin_lock(&con->writequeue_lock);
852 list_for_each_safe(list, temp, &con->writequeue) {
853 struct writequeue_entry *e =
854 list_entry(list, struct writequeue_entry, list);
855 list_del(&e->list);
856 free_entry(e);
857 }
858 spin_unlock(&con->writequeue_lock);
859}
860
861/* Called from recovery when it knows that a node has
862 left the cluster */
863int dlm_lowcomms_close(int nodeid)
864{
865 struct connection *con;
866
867 if (!connections)
868 goto out;
869
870 log_print("closing connection to node %d", nodeid);
871 con = nodeid2con(nodeid, 0);
872 if (con) {
873 clean_one_writequeue(con);
874 close_connection(con, true);
875 atomic_set(&con->waiting_requests, 0);
876 }
877 return 0;
878
879out:
880 return -1;
881}
882
883/* API send message call, may queue the request */
884/* N.B. This is the old interface - use the new one for new calls */
885int lowcomms_send_message(int nodeid, char *buf, int len, gfp_t allocation)
886{
887 struct writequeue_entry *e;
888 char *b;
889
890 e = dlm_lowcomms_get_buffer(nodeid, len, allocation, &b);
891 if (e) {
892 memcpy(b, buf, len);
893 dlm_lowcomms_commit_buffer(e);
894 return 0;
895 }
896 return -ENOBUFS;
897}
898
899/* Look for activity on active sockets */
900static void process_sockets(void)
901{
902 struct list_head *list;
903 struct list_head *temp;
904 int count = 0;
905
906 spin_lock_bh(&read_sockets_lock);
907 list_for_each_safe(list, temp, &read_sockets) {
908
909 struct connection *con =
910 list_entry(list, struct connection, read_list);
911 list_del(&con->read_list);
912 clear_bit(CF_READ_PENDING, &con->flags);
913
914 spin_unlock_bh(&read_sockets_lock);
915
916 /* This can reach zero if we are processing requests
917 * as they come in.
918 */
919 if (atomic_read(&con->waiting_requests) == 0) {
920 spin_lock_bh(&read_sockets_lock);
921 continue;
922 }
923
924 do {
925 con->rx_action(con);
926
927 /* Don't starve out everyone else */
928 if (++count >= MAX_RX_MSG_COUNT) {
929 cond_resched();
930 count = 0;
931 }
932
933 } while (!atomic_dec_and_test(&con->waiting_requests) &&
934 !kthread_should_stop());
935
936 spin_lock_bh(&read_sockets_lock);
937 }
938 spin_unlock_bh(&read_sockets_lock);
939}
940
941/* Try to send any messages that are pending
942 */
943static void process_output_queue(void)
944{
945 struct list_head *list;
946 struct list_head *temp;
947
948 spin_lock_bh(&write_sockets_lock);
949 list_for_each_safe(list, temp, &write_sockets) {
950 struct connection *con =
951 list_entry(list, struct connection, write_list);
952 clear_bit(CF_WRITE_PENDING, &con->flags);
953 list_del(&con->write_list);
954
955 spin_unlock_bh(&write_sockets_lock);
956 send_to_sock(con);
957 spin_lock_bh(&write_sockets_lock);
958 }
959 spin_unlock_bh(&write_sockets_lock);
960}
961
962static void process_state_queue(void)
963{
964 struct list_head *list;
965 struct list_head *temp;
966
967 spin_lock_bh(&state_sockets_lock);
968 list_for_each_safe(list, temp, &state_sockets) {
969 struct connection *con =
970 list_entry(list, struct connection, state_list);
971 list_del(&con->state_list);
972 clear_bit(CF_CONNECT_PENDING, &con->flags);
973 spin_unlock_bh(&state_sockets_lock);
974
975 connect_to_sock(con);
976 spin_lock_bh(&state_sockets_lock);
977 }
978 spin_unlock_bh(&state_sockets_lock);
979}
980
981
982/* Discard all entries on the write queues */
983static void clean_writequeues(void)
984{
985 int nodeid;
986
987 for (nodeid = 1; nodeid < conn_array_size; nodeid++) {
988 struct connection *con = nodeid2con(nodeid, 0);
989
990 if (con)
991 clean_one_writequeue(con);
992 }
993}
994
995static int read_list_empty(void)
996{
997 int status;
998
999 spin_lock_bh(&read_sockets_lock);
1000 status = list_empty(&read_sockets);
1001 spin_unlock_bh(&read_sockets_lock);
1002
1003 return status;
1004}
1005
1006/* DLM Transport comms receive daemon */
1007static int dlm_recvd(void *data)
1008{
1009 init_waitqueue_entry(&lowcomms_recv_waitq_head, current);
1010 add_wait_queue(&lowcomms_recv_waitq, &lowcomms_recv_waitq_head);
1011
1012 while (!kthread_should_stop()) {
1013 set_current_state(TASK_INTERRUPTIBLE);
1014 if (read_list_empty())
1015 cond_resched();
1016 set_current_state(TASK_RUNNING);
1017
1018 process_sockets();
1019 }
1020
1021 return 0;
1022}
1023
1024static int write_and_state_lists_empty(void)
1025{
1026 int status;
1027
1028 spin_lock_bh(&write_sockets_lock);
1029 status = list_empty(&write_sockets);
1030 spin_unlock_bh(&write_sockets_lock);
1031
1032 spin_lock_bh(&state_sockets_lock);
1033 if (list_empty(&state_sockets) == 0)
1034 status = 0;
1035 spin_unlock_bh(&state_sockets_lock);
1036
1037 return status;
1038}
1039
1040/* DLM Transport send daemon */
1041static int dlm_sendd(void *data)
1042{
1043 init_waitqueue_entry(&lowcomms_send_waitq_head, current);
1044 add_wait_queue(&lowcomms_send_waitq, &lowcomms_send_waitq_head);
1045
1046 while (!kthread_should_stop()) {
1047 set_current_state(TASK_INTERRUPTIBLE);
1048 if (write_and_state_lists_empty())
1049 cond_resched();
1050 set_current_state(TASK_RUNNING);
1051
1052 process_state_queue();
1053 process_output_queue();
1054 }
1055
1056 return 0;
1057}
1058
1059static void daemons_stop(void)
1060{
1061 kthread_stop(recv_task);
1062 kthread_stop(send_task);
1063}
1064
1065static int daemons_start(void)
1066{
1067 struct task_struct *p;
1068 int error;
1069
1070 p = kthread_run(dlm_recvd, NULL, "dlm_recvd");
1071 error = IS_ERR(p);
1072 if (error) {
1073 log_print("can't start dlm_recvd %d", error);
1074 return error;
1075 }
1076 recv_task = p;
1077
1078 p = kthread_run(dlm_sendd, NULL, "dlm_sendd");
1079 error = IS_ERR(p);
1080 if (error) {
1081 log_print("can't start dlm_sendd %d", error);
1082 kthread_stop(recv_task);
1083 return error;
1084 }
1085 send_task = p;
1086
1087 return 0;
1088}
1089
1090/*
1091 * Return the largest buffer size we can cope with.
1092 */
1093int lowcomms_max_buffer_size(void)
1094{
1095 return PAGE_CACHE_SIZE;
1096}
1097
1098void dlm_lowcomms_stop(void)
1099{
1100 int i;
1101
1102 /* Set all the flags to prevent any
1103 socket activity.
1104 */
1105 for (i = 0; i < conn_array_size; i++) {
1106 if (connections[i])
1107 connections[i]->flags |= 0xFF;
1108 }
1109
1110 daemons_stop();
1111 clean_writequeues();
1112
1113 for (i = 0; i < conn_array_size; i++) {
1114 if (connections[i]) {
1115 close_connection(connections[i], true);
1116 if (connections[i]->othercon)
1117 kmem_cache_free(con_cache, connections[i]->othercon);
1118 kmem_cache_free(con_cache, connections[i]);
1119 }
1120 }
1121
1122 kfree(connections);
1123 connections = NULL;
1124
1125 kmem_cache_destroy(con_cache);
1126}
1127
1128/* This is quite likely to sleep... */
1129int dlm_lowcomms_start(void)
1130{
1131 int error = 0;
1132
1133 error = -ENOMEM;
1134 connections = kzalloc(sizeof(struct connection *) *
1135 NODE_INCREMENT, GFP_KERNEL);
1136 if (!connections)
1137 goto out;
1138
1139 conn_array_size = NODE_INCREMENT;
1140
1141 if (dlm_our_addr(&dlm_local_addr, 0)) {
1142 log_print("no local IP address has been set");
1143 goto fail_free_conn;
1144 }
1145 if (!dlm_our_addr(&dlm_local_addr, 1)) {
1146 log_print("This dlm comms module does not support multi-homed clustering");
1147 goto fail_free_conn;
1148 }
1149
1150 con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
1151 __alignof__(struct connection), 0,
1152 NULL, NULL);
1153 if (!con_cache)
1154 goto fail_free_conn;
1155
1156
1157 /* Start listening */
1158 error = listen_for_all();
1159 if (error)
1160 goto fail_unlisten;
1161
1162 error = daemons_start();
1163 if (error)
1164 goto fail_unlisten;
1165
1166 return 0;
1167
1168fail_unlisten:
1169 close_connection(connections[0], false);
1170 kmem_cache_free(con_cache, connections[0]);
1171 kmem_cache_destroy(con_cache);
1172
1173fail_free_conn:
1174 kfree(connections);
1175
1176out:
1177 return error;
1178}
1179
1180/*
1181 * Overrides for Emacs so that we follow Linus's tabbing style.
1182 * Emacs will notice this stuff at the end of the file and automatically
1183 * adjust the settings for this buffer only. This must remain at the end
1184 * of the file.
1185 * ---------------------------------------------------------------------------
1186 * Local variables:
1187 * c-file-style: "linux"
1188 * End:
1189 */
generated by cgit v1.2.2 (git 2.25.0) at 2024-11-30 15:15:20 -0500