aboutsummaryrefslogtreecommitdiffstats
path: root/fs/dlm/lowcomms-tcp.c
diff options
context:
space:
mode:
authorPatrick Caulfield <pcaulfie@redhat.com>2007-04-17 10:39:57 -0400
committerSteven Whitehouse <swhiteho@redhat.com>2007-05-01 04:11:23 -0400
commit6ed7257b46709e87d79ac2b6b819b7e0c9184998 (patch)
tree502f68849175f8fb52bb141501df2df9efc8e06c /fs/dlm/lowcomms-tcp.c
parentfc7c44f03d95f20b5446d06f5bb9605cddd53203 (diff)
[DLM] Consolidate transport protocols
This patch consolidates the TCP & SCTP protocols for the DLM into a single file and makes it switchable at run-time (well, at least before the DLM actually starts up!) For RHEL5 this patch requires Neil Horman's patch that expands the in-kernel socket API but that has already been twice ACKed so it should be OK. The patch adds a new lowcomms.c file that replaces the existing lowcomms-sctp.c & lowcomms-tcp.c files. Signed-off-By: Patrick Caulfield <pcaulfie@redhat.com> Signed-off-by: Steven Whitehouse <swhiteho@redhat.com>
Diffstat (limited to 'fs/dlm/lowcomms-tcp.c')
-rw-r--r--fs/dlm/lowcomms-tcp.c1006
1 files changed, 0 insertions, 1006 deletions
diff --git a/fs/dlm/lowcomms-tcp.c b/fs/dlm/lowcomms-tcp.c
deleted file mode 100644
index 919e92a6aebb..000000000000
--- a/fs/dlm/lowcomms-tcp.c
+++ /dev/null
@@ -1,1006 +0,0 @@
1/******************************************************************************
2*******************************************************************************
3**
4** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5** Copyright (C) 2004-2007 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 mutex sock_mutex;
100 unsigned long flags; /* bit 1,2 = We are on the read/write lists */
101#define CF_READ_PENDING 1
102#define CF_WRITE_PENDING 2
103#define CF_CONNECT_PENDING 3
104#define CF_IS_OTHERCON 4
105 struct list_head writequeue; /* List of outgoing writequeue_entries */
106 struct list_head listenlist; /* List of allocated listening sockets */
107 spinlock_t writequeue_lock;
108 int (*rx_action) (struct connection *); /* What to do when active */
109 struct page *rx_page;
110 struct cbuf cb;
111 int retries;
112#define MAX_CONNECT_RETRIES 3
113 struct connection *othercon;
114 struct work_struct rwork; /* Receive workqueue */
115 struct work_struct swork; /* Send workqueue */
116};
117#define sock2con(x) ((struct connection *)(x)->sk_user_data)
118
119/* An entry waiting to be sent */
120struct writequeue_entry {
121 struct list_head list;
122 struct page *page;
123 int offset;
124 int len;
125 int end;
126 int users;
127 struct connection *con;
128};
129
130static struct sockaddr_storage dlm_local_addr;
131
132/* Work queues */
133static struct workqueue_struct *recv_workqueue;
134static struct workqueue_struct *send_workqueue;
135
136/* An array of pointers to connections, indexed by NODEID */
137static struct connection **connections;
138static DECLARE_MUTEX(connections_lock);
139static struct kmem_cache *con_cache;
140static int conn_array_size;
141
142static void process_recv_sockets(struct work_struct *work);
143static void process_send_sockets(struct work_struct *work);
144
145static struct connection *nodeid2con(int nodeid, gfp_t allocation)
146{
147 struct connection *con = NULL;
148
149 down(&connections_lock);
150 if (nodeid >= conn_array_size) {
151 int new_size = nodeid + NODE_INCREMENT;
152 struct connection **new_conns;
153
154 new_conns = kzalloc(sizeof(struct connection *) *
155 new_size, allocation);
156 if (!new_conns)
157 goto finish;
158
159 memcpy(new_conns, connections, sizeof(struct connection *) * conn_array_size);
160 conn_array_size = new_size;
161 kfree(connections);
162 connections = new_conns;
163
164 }
165
166 con = connections[nodeid];
167 if (con == NULL && allocation) {
168 con = kmem_cache_zalloc(con_cache, allocation);
169 if (!con)
170 goto finish;
171
172 con->nodeid = nodeid;
173 mutex_init(&con->sock_mutex);
174 INIT_LIST_HEAD(&con->writequeue);
175 spin_lock_init(&con->writequeue_lock);
176 INIT_WORK(&con->swork, process_send_sockets);
177 INIT_WORK(&con->rwork, process_recv_sockets);
178
179 connections[nodeid] = con;
180 }
181
182finish:
183 up(&connections_lock);
184 return con;
185}
186
187/* Data available on socket or listen socket received a connect */
188static void lowcomms_data_ready(struct sock *sk, int count_unused)
189{
190 struct connection *con = sock2con(sk);
191
192 if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
193 queue_work(recv_workqueue, &con->rwork);
194}
195
196static void lowcomms_write_space(struct sock *sk)
197{
198 struct connection *con = sock2con(sk);
199
200 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
201 queue_work(send_workqueue, &con->swork);
202}
203
204static inline void lowcomms_connect_sock(struct connection *con)
205{
206 if (!test_and_set_bit(CF_CONNECT_PENDING, &con->flags))
207 queue_work(send_workqueue, &con->swork);
208}
209
210static void lowcomms_state_change(struct sock *sk)
211{
212 if (sk->sk_state == TCP_ESTABLISHED)
213 lowcomms_write_space(sk);
214}
215
216/* Make a socket active */
217static int add_sock(struct socket *sock, struct connection *con)
218{
219 con->sock = sock;
220
221 /* Install a data_ready callback */
222 con->sock->sk->sk_data_ready = lowcomms_data_ready;
223 con->sock->sk->sk_write_space = lowcomms_write_space;
224 con->sock->sk->sk_state_change = lowcomms_state_change;
225
226 return 0;
227}
228
229/* Add the port number to an IP6 or 4 sockaddr and return the address
230 length */
231static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
232 int *addr_len)
233{
234 saddr->ss_family = dlm_local_addr.ss_family;
235 if (saddr->ss_family == AF_INET) {
236 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
237 in4_addr->sin_port = cpu_to_be16(port);
238 *addr_len = sizeof(struct sockaddr_in);
239 } else {
240 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
241 in6_addr->sin6_port = cpu_to_be16(port);
242 *addr_len = sizeof(struct sockaddr_in6);
243 }
244}
245
246/* Close a remote connection and tidy up */
247static void close_connection(struct connection *con, bool and_other)
248{
249 mutex_lock(&con->sock_mutex);
250
251 if (con->sock) {
252 sock_release(con->sock);
253 con->sock = NULL;
254 }
255 if (con->othercon && and_other) {
256 /* Will only re-enter once. */
257 close_connection(con->othercon, false);
258 }
259 if (con->rx_page) {
260 __free_page(con->rx_page);
261 con->rx_page = NULL;
262 }
263 con->retries = 0;
264 mutex_unlock(&con->sock_mutex);
265}
266
267/* Data received from remote end */
268static int receive_from_sock(struct connection *con)
269{
270 int ret = 0;
271 struct msghdr msg = {};
272 struct kvec iov[2];
273 unsigned len;
274 int r;
275 int call_again_soon = 0;
276 int nvec;
277
278 mutex_lock(&con->sock_mutex);
279
280 if (con->sock == NULL) {
281 ret = -EAGAIN;
282 goto out_close;
283 }
284
285 if (con->rx_page == NULL) {
286 /*
287 * This doesn't need to be atomic, but I think it should
288 * improve performance if it is.
289 */
290 con->rx_page = alloc_page(GFP_ATOMIC);
291 if (con->rx_page == NULL)
292 goto out_resched;
293 cbuf_init(&con->cb, PAGE_CACHE_SIZE);
294 }
295
296 /*
297 * iov[0] is the bit of the circular buffer between the current end
298 * point (cb.base + cb.len) and the end of the buffer.
299 */
300 iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
301 iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
302 iov[1].iov_len = 0;
303 nvec = 1;
304
305 /*
306 * iov[1] is the bit of the circular buffer between the start of the
307 * buffer and the start of the currently used section (cb.base)
308 */
309 if (cbuf_data(&con->cb) >= con->cb.base) {
310 iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);
311 iov[1].iov_len = con->cb.base;
312 iov[1].iov_base = page_address(con->rx_page);
313 nvec = 2;
314 }
315 len = iov[0].iov_len + iov[1].iov_len;
316
317 r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,
318 MSG_DONTWAIT | MSG_NOSIGNAL);
319
320 if (ret <= 0)
321 goto out_close;
322
323 if (ret == len)
324 call_again_soon = 1;
325 cbuf_add(&con->cb, ret);
326 ret = dlm_process_incoming_buffer(con->nodeid,
327 page_address(con->rx_page),
328 con->cb.base, con->cb.len,
329 PAGE_CACHE_SIZE);
330 if (ret == -EBADMSG) {
331 printk(KERN_INFO "dlm: lowcomms: addr=%p, base=%u, len=%u, "
332 "iov_len=%u, iov_base[0]=%p, read=%d\n",
333 page_address(con->rx_page), con->cb.base, con->cb.len,
334 len, iov[0].iov_base, r);
335 }
336 if (ret < 0)
337 goto out_close;
338 cbuf_eat(&con->cb, ret);
339
340 if (cbuf_empty(&con->cb) && !call_again_soon) {
341 __free_page(con->rx_page);
342 con->rx_page = NULL;
343 }
344
345 if (call_again_soon)
346 goto out_resched;
347 mutex_unlock(&con->sock_mutex);
348 return 0;
349
350out_resched:
351 if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
352 queue_work(recv_workqueue, &con->rwork);
353 mutex_unlock(&con->sock_mutex);
354 return -EAGAIN;
355
356out_close:
357 mutex_unlock(&con->sock_mutex);
358 if (ret != -EAGAIN && !test_bit(CF_IS_OTHERCON, &con->flags)) {
359 close_connection(con, false);
360 /* Reconnect when there is something to send */
361 }
362 /* Don't return success if we really got EOF */
363 if (ret == 0)
364 ret = -EAGAIN;
365
366 return ret;
367}
368
369/* Listening socket is busy, accept a connection */
370static int accept_from_sock(struct connection *con)
371{
372 int result;
373 struct sockaddr_storage peeraddr;
374 struct socket *newsock;
375 int len;
376 int nodeid;
377 struct connection *newcon;
378 struct connection *addcon;
379
380 memset(&peeraddr, 0, sizeof(peeraddr));
381 result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM,
382 IPPROTO_TCP, &newsock);
383 if (result < 0)
384 return -ENOMEM;
385
386 mutex_lock_nested(&con->sock_mutex, 0);
387
388 result = -ENOTCONN;
389 if (con->sock == NULL)
390 goto accept_err;
391
392 newsock->type = con->sock->type;
393 newsock->ops = con->sock->ops;
394
395 result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK);
396 if (result < 0)
397 goto accept_err;
398
399 /* Get the connected socket's peer */
400 memset(&peeraddr, 0, sizeof(peeraddr));
401 if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr,
402 &len, 2)) {
403 result = -ECONNABORTED;
404 goto accept_err;
405 }
406
407 /* Get the new node's NODEID */
408 make_sockaddr(&peeraddr, 0, &len);
409 if (dlm_addr_to_nodeid(&peeraddr, &nodeid)) {
410 printk("dlm: connect from non cluster node\n");
411 sock_release(newsock);
412 mutex_unlock(&con->sock_mutex);
413 return -1;
414 }
415
416 log_print("got connection from %d", nodeid);
417
418 /* Check to see if we already have a connection to this node. This
419 * could happen if the two nodes initiate a connection at roughly
420 * the same time and the connections cross on the wire.
421 * TEMPORARY FIX:
422 * In this case we store the incoming one in "othercon"
423 */
424 newcon = nodeid2con(nodeid, GFP_KERNEL);
425 if (!newcon) {
426 result = -ENOMEM;
427 goto accept_err;
428 }
429 mutex_lock_nested(&newcon->sock_mutex, 1);
430 if (newcon->sock) {
431 struct connection *othercon = newcon->othercon;
432
433 if (!othercon) {
434 othercon = kmem_cache_zalloc(con_cache, GFP_KERNEL);
435 if (!othercon) {
436 printk("dlm: failed to allocate incoming socket\n");
437 mutex_unlock(&newcon->sock_mutex);
438 result = -ENOMEM;
439 goto accept_err;
440 }
441 othercon->nodeid = nodeid;
442 othercon->rx_action = receive_from_sock;
443 mutex_init(&othercon->sock_mutex);
444 INIT_WORK(&othercon->swork, process_send_sockets);
445 INIT_WORK(&othercon->rwork, process_recv_sockets);
446 set_bit(CF_IS_OTHERCON, &othercon->flags);
447 newcon->othercon = othercon;
448 }
449 othercon->sock = newsock;
450 newsock->sk->sk_user_data = othercon;
451 add_sock(newsock, othercon);
452 addcon = othercon;
453 }
454 else {
455 newsock->sk->sk_user_data = newcon;
456 newcon->rx_action = receive_from_sock;
457 add_sock(newsock, newcon);
458 addcon = newcon;
459 }
460
461 mutex_unlock(&newcon->sock_mutex);
462
463 /*
464 * Add it to the active queue in case we got data
465 * beween processing the accept adding the socket
466 * to the read_sockets list
467 */
468 if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
469 queue_work(recv_workqueue, &addcon->rwork);
470 mutex_unlock(&con->sock_mutex);
471
472 return 0;
473
474accept_err:
475 mutex_unlock(&con->sock_mutex);
476 sock_release(newsock);
477
478 if (result != -EAGAIN)
479 printk("dlm: error accepting connection from node: %d\n", result);
480 return result;
481}
482
483/* Connect a new socket to its peer */
484static void connect_to_sock(struct connection *con)
485{
486 int result = -EHOSTUNREACH;
487 struct sockaddr_storage saddr;
488 int addr_len;
489 struct socket *sock;
490
491 if (con->nodeid == 0) {
492 log_print("attempt to connect sock 0 foiled");
493 return;
494 }
495
496 mutex_lock(&con->sock_mutex);
497 if (con->retries++ > MAX_CONNECT_RETRIES)
498 goto out;
499
500 /* Some odd races can cause double-connects, ignore them */
501 if (con->sock) {
502 result = 0;
503 goto out;
504 }
505
506 /* Create a socket to communicate with */
507 result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM,
508 IPPROTO_TCP, &sock);
509 if (result < 0)
510 goto out_err;
511
512 memset(&saddr, 0, sizeof(saddr));
513 if (dlm_nodeid_to_addr(con->nodeid, &saddr))
514 goto out_err;
515
516 sock->sk->sk_user_data = con;
517 con->rx_action = receive_from_sock;
518
519 make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
520
521 add_sock(sock, con);
522
523 log_print("connecting to %d", con->nodeid);
524 result =
525 sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
526 O_NONBLOCK);
527 if (result == -EINPROGRESS)
528 result = 0;
529 if (result == 0)
530 goto out;
531
532out_err:
533 if (con->sock) {
534 sock_release(con->sock);
535 con->sock = NULL;
536 }
537 /*
538 * Some errors are fatal and this list might need adjusting. For other
539 * errors we try again until the max number of retries is reached.
540 */
541 if (result != -EHOSTUNREACH && result != -ENETUNREACH &&
542 result != -ENETDOWN && result != EINVAL
543 && result != -EPROTONOSUPPORT) {
544 lowcomms_connect_sock(con);
545 result = 0;
546 }
547out:
548 mutex_unlock(&con->sock_mutex);
549 return;
550}
551
552static struct socket *create_listen_sock(struct connection *con,
553 struct sockaddr_storage *saddr)
554{
555 struct socket *sock = NULL;
556 mm_segment_t fs;
557 int result = 0;
558 int one = 1;
559 int addr_len;
560
561 if (dlm_local_addr.ss_family == AF_INET)
562 addr_len = sizeof(struct sockaddr_in);
563 else
564 addr_len = sizeof(struct sockaddr_in6);
565
566 /* Create a socket to communicate with */
567 result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM, IPPROTO_TCP, &sock);
568 if (result < 0) {
569 printk("dlm: Can't create listening comms socket\n");
570 goto create_out;
571 }
572
573 fs = get_fs();
574 set_fs(get_ds());
575 result = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
576 (char *)&one, sizeof(one));
577 set_fs(fs);
578 if (result < 0) {
579 printk("dlm: Failed to set SO_REUSEADDR on socket: result=%d\n",
580 result);
581 }
582 sock->sk->sk_user_data = con;
583 con->rx_action = accept_from_sock;
584 con->sock = sock;
585
586 /* Bind to our port */
587 make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
588 result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
589 if (result < 0) {
590 printk("dlm: Can't bind to port %d\n", dlm_config.ci_tcp_port);
591 sock_release(sock);
592 sock = NULL;
593 con->sock = NULL;
594 goto create_out;
595 }
596
597 fs = get_fs();
598 set_fs(get_ds());
599
600 result = sock_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
601 (char *)&one, sizeof(one));
602 set_fs(fs);
603 if (result < 0) {
604 printk("dlm: Set keepalive failed: %d\n", result);
605 }
606
607 result = sock->ops->listen(sock, 5);
608 if (result < 0) {
609 printk("dlm: Can't listen on port %d\n", dlm_config.ci_tcp_port);
610 sock_release(sock);
611 sock = NULL;
612 goto create_out;
613 }
614
615create_out:
616 return sock;
617}
618
619
620/* Listen on all interfaces */
621static int listen_for_all(void)
622{
623 struct socket *sock = NULL;
624 struct connection *con = nodeid2con(0, GFP_KERNEL);
625 int result = -EINVAL;
626
627 /* We don't support multi-homed hosts */
628 set_bit(CF_IS_OTHERCON, &con->flags);
629
630 sock = create_listen_sock(con, &dlm_local_addr);
631 if (sock) {
632 add_sock(sock, con);
633 result = 0;
634 }
635 else {
636 result = -EADDRINUSE;
637 }
638
639 return result;
640}
641
642
643
644static struct writequeue_entry *new_writequeue_entry(struct connection *con,
645 gfp_t allocation)
646{
647 struct writequeue_entry *entry;
648
649 entry = kmalloc(sizeof(struct writequeue_entry), allocation);
650 if (!entry)
651 return NULL;
652
653 entry->page = alloc_page(allocation);
654 if (!entry->page) {
655 kfree(entry);
656 return NULL;
657 }
658
659 entry->offset = 0;
660 entry->len = 0;
661 entry->end = 0;
662 entry->users = 0;
663 entry->con = con;
664
665 return entry;
666}
667
668void *dlm_lowcomms_get_buffer(int nodeid, int len,
669 gfp_t allocation, char **ppc)
670{
671 struct connection *con;
672 struct writequeue_entry *e;
673 int offset = 0;
674 int users = 0;
675
676 con = nodeid2con(nodeid, allocation);
677 if (!con)
678 return NULL;
679
680 spin_lock(&con->writequeue_lock);
681 e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
682 if ((&e->list == &con->writequeue) ||
683 (PAGE_CACHE_SIZE - e->end < len)) {
684 e = NULL;
685 } else {
686 offset = e->end;
687 e->end += len;
688 users = e->users++;
689 }
690 spin_unlock(&con->writequeue_lock);
691
692 if (e) {
693 got_one:
694 if (users == 0)
695 kmap(e->page);
696 *ppc = page_address(e->page) + offset;
697 return e;
698 }
699
700 e = new_writequeue_entry(con, allocation);
701 if (e) {
702 spin_lock(&con->writequeue_lock);
703 offset = e->end;
704 e->end += len;
705 users = e->users++;
706 list_add_tail(&e->list, &con->writequeue);
707 spin_unlock(&con->writequeue_lock);
708 goto got_one;
709 }
710 return NULL;
711}
712
713void dlm_lowcomms_commit_buffer(void *mh)
714{
715 struct writequeue_entry *e = (struct writequeue_entry *)mh;
716 struct connection *con = e->con;
717 int users;
718
719 spin_lock(&con->writequeue_lock);
720 users = --e->users;
721 if (users)
722 goto out;
723 e->len = e->end - e->offset;
724 kunmap(e->page);
725 spin_unlock(&con->writequeue_lock);
726
727 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {
728 queue_work(send_workqueue, &con->swork);
729 }
730 return;
731
732out:
733 spin_unlock(&con->writequeue_lock);
734 return;
735}
736
737static void free_entry(struct writequeue_entry *e)
738{
739 __free_page(e->page);
740 kfree(e);
741}
742
743/* Send a message */
744static void send_to_sock(struct connection *con)
745{
746 int ret = 0;
747 ssize_t(*sendpage) (struct socket *, struct page *, int, size_t, int);
748 const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
749 struct writequeue_entry *e;
750 int len, offset;
751
752 mutex_lock(&con->sock_mutex);
753 if (con->sock == NULL)
754 goto out_connect;
755
756 sendpage = con->sock->ops->sendpage;
757
758 spin_lock(&con->writequeue_lock);
759 for (;;) {
760 e = list_entry(con->writequeue.next, struct writequeue_entry,
761 list);
762 if ((struct list_head *) e == &con->writequeue)
763 break;
764
765 len = e->len;
766 offset = e->offset;
767 BUG_ON(len == 0 && e->users == 0);
768 spin_unlock(&con->writequeue_lock);
769 kmap(e->page);
770
771 ret = 0;
772 if (len) {
773 ret = sendpage(con->sock, e->page, offset, len,
774 msg_flags);
775 if (ret == -EAGAIN || ret == 0)
776 goto out;
777 if (ret <= 0)
778 goto send_error;
779 }
780 else {
781 /* Don't starve people filling buffers */
782 cond_resched();
783 }
784
785 spin_lock(&con->writequeue_lock);
786 e->offset += ret;
787 e->len -= ret;
788
789 if (e->len == 0 && e->users == 0) {
790 list_del(&e->list);
791 kunmap(e->page);
792 free_entry(e);
793 continue;
794 }
795 }
796 spin_unlock(&con->writequeue_lock);
797out:
798 mutex_unlock(&con->sock_mutex);
799 return;
800
801send_error:
802 mutex_unlock(&con->sock_mutex);
803 close_connection(con, false);
804 lowcomms_connect_sock(con);
805 return;
806
807out_connect:
808 mutex_unlock(&con->sock_mutex);
809 connect_to_sock(con);
810 return;
811}
812
813static void clean_one_writequeue(struct connection *con)
814{
815 struct list_head *list;
816 struct list_head *temp;
817
818 spin_lock(&con->writequeue_lock);
819 list_for_each_safe(list, temp, &con->writequeue) {
820 struct writequeue_entry *e =
821 list_entry(list, struct writequeue_entry, list);
822 list_del(&e->list);
823 free_entry(e);
824 }
825 spin_unlock(&con->writequeue_lock);
826}
827
828/* Called from recovery when it knows that a node has
829 left the cluster */
830int dlm_lowcomms_close(int nodeid)
831{
832 struct connection *con;
833
834 if (!connections)
835 goto out;
836
837 log_print("closing connection to node %d", nodeid);
838 con = nodeid2con(nodeid, 0);
839 if (con) {
840 clean_one_writequeue(con);
841 close_connection(con, true);
842 }
843 return 0;
844
845out:
846 return -1;
847}
848
849/* Look for activity on active sockets */
850static void process_recv_sockets(struct work_struct *work)
851{
852 struct connection *con = container_of(work, struct connection, rwork);
853 int err;
854
855 clear_bit(CF_READ_PENDING, &con->flags);
856 do {
857 err = con->rx_action(con);
858 } while (!err);
859}
860
861
862static void process_send_sockets(struct work_struct *work)
863{
864 struct connection *con = container_of(work, struct connection, swork);
865
866 if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
867 connect_to_sock(con);
868 }
869
870 clear_bit(CF_WRITE_PENDING, &con->flags);
871 send_to_sock(con);
872}
873
874
875/* Discard all entries on the write queues */
876static void clean_writequeues(void)
877{
878 int nodeid;
879
880 for (nodeid = 1; nodeid < conn_array_size; nodeid++) {
881 struct connection *con = nodeid2con(nodeid, 0);
882
883 if (con)
884 clean_one_writequeue(con);
885 }
886}
887
888static void work_stop(void)
889{
890 destroy_workqueue(recv_workqueue);
891 destroy_workqueue(send_workqueue);
892}
893
894static int work_start(void)
895{
896 int error;
897 recv_workqueue = create_workqueue("dlm_recv");
898 error = IS_ERR(recv_workqueue);
899 if (error) {
900 log_print("can't start dlm_recv %d", error);
901 return error;
902 }
903
904 send_workqueue = create_singlethread_workqueue("dlm_send");
905 error = IS_ERR(send_workqueue);
906 if (error) {
907 log_print("can't start dlm_send %d", error);
908 destroy_workqueue(recv_workqueue);
909 return error;
910 }
911
912 return 0;
913}
914
915void dlm_lowcomms_stop(void)
916{
917 int i;
918
919 /* Set all the flags to prevent any
920 socket activity.
921 */
922 for (i = 0; i < conn_array_size; i++) {
923 if (connections[i])
924 connections[i]->flags |= 0xFF;
925 }
926
927 work_stop();
928 clean_writequeues();
929
930 for (i = 0; i < conn_array_size; i++) {
931 if (connections[i]) {
932 close_connection(connections[i], true);
933 if (connections[i]->othercon)
934 kmem_cache_free(con_cache, connections[i]->othercon);
935 kmem_cache_free(con_cache, connections[i]);
936 }
937 }
938
939 kfree(connections);
940 connections = NULL;
941
942 kmem_cache_destroy(con_cache);
943}
944
945/* This is quite likely to sleep... */
946int dlm_lowcomms_start(void)
947{
948 int error = 0;
949
950 error = -ENOMEM;
951 connections = kzalloc(sizeof(struct connection *) *
952 NODE_INCREMENT, GFP_KERNEL);
953 if (!connections)
954 goto out;
955
956 conn_array_size = NODE_INCREMENT;
957
958 if (dlm_our_addr(&dlm_local_addr, 0)) {
959 log_print("no local IP address has been set");
960 goto fail_free_conn;
961 }
962 if (!dlm_our_addr(&dlm_local_addr, 1)) {
963 log_print("This dlm comms module does not support multi-homed clustering");
964 goto fail_free_conn;
965 }
966
967 con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
968 __alignof__(struct connection), 0,
969 NULL, NULL);
970 if (!con_cache)
971 goto fail_free_conn;
972
973
974 /* Start listening */
975 error = listen_for_all();
976 if (error)
977 goto fail_unlisten;
978
979 error = work_start();
980 if (error)
981 goto fail_unlisten;
982
983 return 0;
984
985fail_unlisten:
986 close_connection(connections[0], false);
987 kmem_cache_free(con_cache, connections[0]);
988 kmem_cache_destroy(con_cache);
989
990fail_free_conn:
991 kfree(connections);
992
993out:
994 return error;
995}
996
997/*
998 * Overrides for Emacs so that we follow Linus's tabbing style.
999 * Emacs will notice this stuff at the end of the file and automatically
1000 * adjust the settings for this buffer only. This must remain at the end
1001 * of the file.
1002 * ---------------------------------------------------------------------------
1003 * Local variables:
1004 * c-file-style: "linux"
1005 * End:
1006 */
generated by cgit v1.2.2 (git 2.25.0) at 2025-03-11 04:48:26 -0400