diff options
author | Patrick Caulfield <pcaulfie@redhat.com> | 2007-04-17 15:39:57 +0100 |
---|---|---|
committer | Steven Whitehouse <swhiteho@redhat.com> | 2007-05-01 09:11:23 +0100 |
commit | 6ed7257b46709e87d79ac2b6b819b7e0c9184998 (patch) | |
tree | 502f68849175f8fb52bb141501df2df9efc8e06c /fs/dlm/lowcomms-tcp.c | |
parent | fc7c44f03d95f20b5446d06f5bb9605cddd53203 (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.c | 1006 |
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 919e92a6aeb..00000000000 --- a/fs/dlm/lowcomms-tcp.c +++ /dev/null @@ -1,1006 +0,0 @@ -/****************************************************************************** -******************************************************************************* -** -** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved. -** Copyright (C) 2004-2007 Red Hat, Inc. All rights reserved. -** -** This copyrighted material is made available to anyone wishing to use, -** modify, copy, or redistribute it subject to the terms and conditions -** of the GNU General Public License v.2. -** -******************************************************************************* -******************************************************************************/ - -/* - * lowcomms.c - * - * This is the "low-level" comms layer. - * - * It is responsible for sending/receiving messages - * from other nodes in the cluster. - * - * Cluster nodes are referred to by their nodeids. nodeids are - * simply 32 bit numbers to the locking module - if they need to - * be expanded for the cluster infrastructure then that is it's - * responsibility. It is this layer's - * responsibility to resolve these into IP address or - * whatever it needs for inter-node communication. - * - * The comms level is two kernel threads that deal mainly with - * the receiving of messages from other nodes and passing them - * up to the mid-level comms layer (which understands the - * message format) for execution by the locking core, and - * a send thread which does all the setting up of connections - * to remote nodes and the sending of data. Threads are not allowed - * to send their own data because it may cause them to wait in times - * of high load. Also, this way, the sending thread can collect together - * messages bound for one node and send them in one block. - * - * I don't see any problem with the recv thread executing the locking - * code on behalf of remote processes as the locking code is - * short, efficient and never waits. - * - */ - - -#include <asm/ioctls.h> -#include <net/sock.h> -#include <net/tcp.h> -#include <linux/pagemap.h> - -#include "dlm_internal.h" -#include "lowcomms.h" -#include "midcomms.h" -#include "config.h" - -struct cbuf { - unsigned int base; - unsigned int len; - unsigned int mask; -}; - -#define NODE_INCREMENT 32 -static void cbuf_add(struct cbuf *cb, int n) -{ - cb->len += n; -} - -static int cbuf_data(struct cbuf *cb) -{ - return ((cb->base + cb->len) & cb->mask); -} - -static void cbuf_init(struct cbuf *cb, int size) -{ - cb->base = cb->len = 0; - cb->mask = size-1; -} - -static void cbuf_eat(struct cbuf *cb, int n) -{ - cb->len -= n; - cb->base += n; - cb->base &= cb->mask; -} - -static bool cbuf_empty(struct cbuf *cb) -{ - return cb->len == 0; -} - -/* Maximum number of incoming messages to process before - doing a cond_resched() -*/ -#define MAX_RX_MSG_COUNT 25 - -struct connection { - struct socket *sock; /* NULL if not connected */ - uint32_t nodeid; /* So we know who we are in the list */ - struct mutex sock_mutex; - unsigned long flags; /* bit 1,2 = We are on the read/write lists */ -#define CF_READ_PENDING 1 -#define CF_WRITE_PENDING 2 -#define CF_CONNECT_PENDING 3 -#define CF_IS_OTHERCON 4 - struct list_head writequeue; /* List of outgoing writequeue_entries */ - struct list_head listenlist; /* List of allocated listening sockets */ - spinlock_t writequeue_lock; - int (*rx_action) (struct connection *); /* What to do when active */ - struct page *rx_page; - struct cbuf cb; - int retries; -#define MAX_CONNECT_RETRIES 3 - struct connection *othercon; - struct work_struct rwork; /* Receive workqueue */ - struct work_struct swork; /* Send workqueue */ -}; -#define sock2con(x) ((struct connection *)(x)->sk_user_data) - -/* An entry waiting to be sent */ -struct writequeue_entry { - struct list_head list; - struct page *page; - int offset; - int len; - int end; - int users; - struct connection *con; -}; - -static struct sockaddr_storage dlm_local_addr; - -/* Work queues */ -static struct workqueue_struct *recv_workqueue; -static struct workqueue_struct *send_workqueue; - -/* An array of pointers to connections, indexed by NODEID */ -static struct connection **connections; -static DECLARE_MUTEX(connections_lock); -static struct kmem_cache *con_cache; -static int conn_array_size; - -static void process_recv_sockets(struct work_struct *work); -static void process_send_sockets(struct work_struct *work); - -static struct connection *nodeid2con(int nodeid, gfp_t allocation) -{ - struct connection *con = NULL; - - down(&connections_lock); - if (nodeid >= conn_array_size) { - int new_size = nodeid + NODE_INCREMENT; - struct connection **new_conns; - - new_conns = kzalloc(sizeof(struct connection *) * - new_size, allocation); - if (!new_conns) - goto finish; - - memcpy(new_conns, connections, sizeof(struct connection *) * conn_array_size); - conn_array_size = new_size; - kfree(connections); - connections = new_conns; - - } - - con = connections[nodeid]; - if (con == NULL && allocation) { - con = kmem_cache_zalloc(con_cache, allocation); - if (!con) - goto finish; - - con->nodeid = nodeid; - mutex_init(&con->sock_mutex); - INIT_LIST_HEAD(&con->writequeue); - spin_lock_init(&con->writequeue_lock); - INIT_WORK(&con->swork, process_send_sockets); - INIT_WORK(&con->rwork, process_recv_sockets); - - connections[nodeid] = con; - } - -finish: - up(&connections_lock); - return con; -} - -/* Data available on socket or listen socket received a connect */ -static void lowcomms_data_ready(struct sock *sk, int count_unused) -{ - struct connection *con = sock2con(sk); - - if (!test_and_set_bit(CF_READ_PENDING, &con->flags)) - queue_work(recv_workqueue, &con->rwork); -} - -static void lowcomms_write_space(struct sock *sk) -{ - struct connection *con = sock2con(sk); - - if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) - queue_work(send_workqueue, &con->swork); -} - -static inline void lowcomms_connect_sock(struct connection *con) -{ - if (!test_and_set_bit(CF_CONNECT_PENDING, &con->flags)) - queue_work(send_workqueue, &con->swork); -} - -static void lowcomms_state_change(struct sock *sk) -{ - if (sk->sk_state == TCP_ESTABLISHED) - lowcomms_write_space(sk); -} - -/* Make a socket active */ -static int add_sock(struct socket *sock, struct connection *con) -{ - con->sock = sock; - - /* Install a data_ready callback */ - con->sock->sk->sk_data_ready = lowcomms_data_ready; - con->sock->sk->sk_write_space = lowcomms_write_space; - con->sock->sk->sk_state_change = lowcomms_state_change; - - return 0; -} - -/* Add the port number to an IP6 or 4 sockaddr and return the address - length */ -static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port, - int *addr_len) -{ - saddr->ss_family = dlm_local_addr.ss_family; - if (saddr->ss_family == AF_INET) { - struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr; - in4_addr->sin_port = cpu_to_be16(port); - *addr_len = sizeof(struct sockaddr_in); - } else { - struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr; - in6_addr->sin6_port = cpu_to_be16(port); - *addr_len = sizeof(struct sockaddr_in6); - } -} - -/* Close a remote connection and tidy up */ -static void close_connection(struct connection *con, bool and_other) -{ - mutex_lock(&con->sock_mutex); - - if (con->sock) { - sock_release(con->sock); - con->sock = NULL; - } - if (con->othercon && and_other) { - /* Will only re-enter once. */ - close_connection(con->othercon, false); - } - if (con->rx_page) { - __free_page(con->rx_page); - con->rx_page = NULL; - } - con->retries = 0; - mutex_unlock(&con->sock_mutex); -} - -/* Data received from remote end */ -static int receive_from_sock(struct connection *con) -{ - int ret = 0; - struct msghdr msg = {}; - struct kvec iov[2]; - unsigned len; - int r; - int call_again_soon = 0; - int nvec; - - mutex_lock(&con->sock_mutex); - - if (con->sock == NULL) { - ret = -EAGAIN; - goto out_close; - } - - if (con->rx_page == NULL) { - /* - * This doesn't need to be atomic, but I think it should - * improve performance if it is. - */ - con->rx_page = alloc_page(GFP_ATOMIC); - if (con->rx_page == NULL) - goto out_resched; - cbuf_init(&con->cb, PAGE_CACHE_SIZE); - } - - /* - * iov[0] is the bit of the circular buffer between the current end - * point (cb.base + cb.len) and the end of the buffer. - */ - iov[0].iov_len = con->cb.base - cbuf_data(&con->cb); - iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb); - iov[1].iov_len = 0; - nvec = 1; - - /* - * iov[1] is the bit of the circular buffer between the start of the - * buffer and the start of the currently used section (cb.base) - */ - if (cbuf_data(&con->cb) >= con->cb.base) { - iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb); - iov[1].iov_len = con->cb.base; - iov[1].iov_base = page_address(con->rx_page); - nvec = 2; - } - len = iov[0].iov_len + iov[1].iov_len; - - r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len, - MSG_DONTWAIT | MSG_NOSIGNAL); - - if (ret <= 0) - goto out_close; - - if (ret == len) - call_again_soon = 1; - cbuf_add(&con->cb, ret); - ret = dlm_process_incoming_buffer(con->nodeid, - page_address(con->rx_page), - con->cb.base, con->cb.len, - PAGE_CACHE_SIZE); - if (ret == -EBADMSG) { - printk(KERN_INFO "dlm: lowcomms: addr=%p, base=%u, len=%u, " - "iov_len=%u, iov_base[0]=%p, read=%d\n", - page_address(con->rx_page), con->cb.base, con->cb.len, - len, iov[0].iov_base, r); - } - if (ret < 0) - goto out_close; - cbuf_eat(&con->cb, ret); - - if (cbuf_empty(&con->cb) && !call_again_soon) { - __free_page(con->rx_page); - con->rx_page = NULL; - } - - if (call_again_soon) - goto out_resched; - mutex_unlock(&con->sock_mutex); - return 0; - -out_resched: - if (!test_and_set_bit(CF_READ_PENDING, &con->flags)) - queue_work(recv_workqueue, &con->rwork); - mutex_unlock(&con->sock_mutex); - return -EAGAIN; - -out_close: - mutex_unlock(&con->sock_mutex); - if (ret != -EAGAIN && !test_bit(CF_IS_OTHERCON, &con->flags)) { - close_connection(con, false); - /* Reconnect when there is something to send */ - } - /* Don't return success if we really got EOF */ - if (ret == 0) - ret = -EAGAIN; - - return ret; -} - -/* Listening socket is busy, accept a connection */ -static int accept_from_sock(struct connection *con) -{ - int result; - struct sockaddr_storage peeraddr; - struct socket *newsock; - int len; - int nodeid; - struct connection *newcon; - struct connection *addcon; - - memset(&peeraddr, 0, sizeof(peeraddr)); - result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM, - IPPROTO_TCP, &newsock); - if (result < 0) - return -ENOMEM; - - mutex_lock_nested(&con->sock_mutex, 0); - - result = -ENOTCONN; - if (con->sock == NULL) - goto accept_err; - - newsock->type = con->sock->type; - newsock->ops = con->sock->ops; - - result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK); - if (result < 0) - goto accept_err; - - /* Get the connected socket's peer */ - memset(&peeraddr, 0, sizeof(peeraddr)); - if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, - &len, 2)) { - result = -ECONNABORTED; - goto accept_err; - } - - /* Get the new node's NODEID */ - make_sockaddr(&peeraddr, 0, &len); - if (dlm_addr_to_nodeid(&peeraddr, &nodeid)) { - printk("dlm: connect from non cluster node\n"); - sock_release(newsock); - mutex_unlock(&con->sock_mutex); - return -1; - } - - log_print("got connection from %d", nodeid); - - /* Check to see if we already have a connection to this node. This - * could happen if the two nodes initiate a connection at roughly - * the same time and the connections cross on the wire. - * TEMPORARY FIX: - * In this case we store the incoming one in "othercon" - */ - newcon = nodeid2con(nodeid, GFP_KERNEL); - if (!newcon) { - result = -ENOMEM; - goto accept_err; - } - mutex_lock_nested(&newcon->sock_mutex, 1); - if (newcon->sock) { - struct connection *othercon = newcon->othercon; - - if (!othercon) { - othercon = kmem_cache_zalloc(con_cache, GFP_KERNEL); - if (!othercon) { - printk("dlm: failed to allocate incoming socket\n"); - mutex_unlock(&newcon->sock_mutex); - result = -ENOMEM; - goto accept_err; - } - othercon->nodeid = nodeid; - othercon->rx_action = receive_from_sock; - mutex_init(&othercon->sock_mutex); - INIT_WORK(&othercon->swork, process_send_sockets); - INIT_WORK(&othercon->rwork, process_recv_sockets); - set_bit(CF_IS_OTHERCON, &othercon->flags); - newcon->othercon = othercon; - } - othercon->sock = newsock; - newsock->sk->sk_user_data = othercon; - add_sock(newsock, othercon); - addcon = othercon; - } - else { - newsock->sk->sk_user_data = newcon; - newcon->rx_action = receive_from_sock; - add_sock(newsock, newcon); - addcon = newcon; - } - - mutex_unlock(&newcon->sock_mutex); - - /* - * Add it to the active queue in case we got data - * beween processing the accept adding the socket - * to the read_sockets list - */ - if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags)) - queue_work(recv_workqueue, &addcon->rwork); - mutex_unlock(&con->sock_mutex); - - return 0; - -accept_err: - mutex_unlock(&con->sock_mutex); - sock_release(newsock); - - if (result != -EAGAIN) - printk("dlm: error accepting connection from node: %d\n", result); - return result; -} - -/* Connect a new socket to its peer */ -static void connect_to_sock(struct connection *con) -{ - int result = -EHOSTUNREACH; - struct sockaddr_storage saddr; - int addr_len; - struct socket *sock; - - if (con->nodeid == 0) { - log_print("attempt to connect sock 0 foiled"); - return; - } - - mutex_lock(&con->sock_mutex); - if (con->retries++ > MAX_CONNECT_RETRIES) - goto out; - - /* Some odd races can cause double-connects, ignore them */ - if (con->sock) { - result = 0; - goto out; - } - - /* Create a socket to communicate with */ - result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM, - IPPROTO_TCP, &sock); - if (result < 0) - goto out_err; - - memset(&saddr, 0, sizeof(saddr)); - if (dlm_nodeid_to_addr(con->nodeid, &saddr)) - goto out_err; - - sock->sk->sk_user_data = con; - con->rx_action = receive_from_sock; - - make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len); - - add_sock(sock, con); - - log_print("connecting to %d", con->nodeid); - result = - sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len, - O_NONBLOCK); - if (result == -EINPROGRESS) - result = 0; - if (result == 0) - goto out; - -out_err: - if (con->sock) { - sock_release(con->sock); - con->sock = NULL; - } - /* - * Some errors are fatal and this list might need adjusting. For other - * errors we try again until the max number of retries is reached. - */ - if (result != -EHOSTUNREACH && result != -ENETUNREACH && - result != -ENETDOWN && result != EINVAL - && result != -EPROTONOSUPPORT) { - lowcomms_connect_sock(con); - result = 0; - } -out: - mutex_unlock(&con->sock_mutex); - return; -} - -static struct socket *create_listen_sock(struct connection *con, - struct sockaddr_storage *saddr) -{ - struct socket *sock = NULL; - mm_segment_t fs; - int result = 0; - int one = 1; - int addr_len; - - if (dlm_local_addr.ss_family == AF_INET) - addr_len = sizeof(struct sockaddr_in); - else - addr_len = sizeof(struct sockaddr_in6); - - /* Create a socket to communicate with */ - result = sock_create_kern(dlm_local_addr.ss_family, SOCK_STREAM, IPPROTO_TCP, &sock); - if (result < 0) { - printk("dlm: Can't create listening comms socket\n"); - goto create_out; - } - - fs = get_fs(); - set_fs(get_ds()); - result = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, - (char *)&one, sizeof(one)); - set_fs(fs); - if (result < 0) { - printk("dlm: Failed to set SO_REUSEADDR on socket: result=%d\n", - result); - } - sock->sk->sk_user_data = con; - con->rx_action = accept_from_sock; - con->sock = sock; - - /* Bind to our port */ - make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len); - result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len); - if (result < 0) { - printk("dlm: Can't bind to port %d\n", dlm_config.ci_tcp_port); - sock_release(sock); - sock = NULL; - con->sock = NULL; - goto create_out; - } - - fs = get_fs(); - set_fs(get_ds()); - - result = sock_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, - (char *)&one, sizeof(one)); - set_fs(fs); - if (result < 0) { - printk("dlm: Set keepalive failed: %d\n", result); - } - - result = sock->ops->listen(sock, 5); - if (result < 0) { - printk("dlm: Can't listen on port %d\n", dlm_config.ci_tcp_port); - sock_release(sock); - sock = NULL; - goto create_out; - } - -create_out: - return sock; -} - - -/* Listen on all interfaces */ -static int listen_for_all(void) -{ - struct socket *sock = NULL; - struct connection *con = nodeid2con(0, GFP_KERNEL); - int result = -EINVAL; - - /* We don't support multi-homed hosts */ - set_bit(CF_IS_OTHERCON, &con->flags); - - sock = create_listen_sock(con, &dlm_local_addr); - if (sock) { - add_sock(sock, con); - result = 0; - } - else { - result = -EADDRINUSE; - } - - return result; -} - - - -static struct writequeue_entry *new_writequeue_entry(struct connection *con, - gfp_t allocation) -{ - struct writequeue_entry *entry; - - entry = kmalloc(sizeof(struct writequeue_entry), allocation); - if (!entry) - return NULL; - - entry->page = alloc_page(allocation); - if (!entry->page) { - kfree(entry); - return NULL; - } - - entry->offset = 0; - entry->len = 0; - entry->end = 0; - entry->users = 0; - entry->con = con; - - return entry; -} - -void *dlm_lowcomms_get_buffer(int nodeid, int len, - gfp_t allocation, char **ppc) -{ - struct connection *con; - struct writequeue_entry *e; - int offset = 0; - int users = 0; - - con = nodeid2con(nodeid, allocation); - if (!con) - return NULL; - - spin_lock(&con->writequeue_lock); - e = list_entry(con->writequeue.prev, struct writequeue_entry, list); - if ((&e->list == &con->writequeue) || - (PAGE_CACHE_SIZE - e->end < len)) { - e = NULL; - } else { - offset = e->end; - e->end += len; - users = e->users++; - } - spin_unlock(&con->writequeue_lock); - - if (e) { - got_one: - if (users == 0) - kmap(e->page); - *ppc = page_address(e->page) + offset; - return e; - } - - e = new_writequeue_entry(con, allocation); - if (e) { - spin_lock(&con->writequeue_lock); - offset = e->end; - e->end += len; - users = e->users++; - list_add_tail(&e->list, &con->writequeue); - spin_unlock(&con->writequeue_lock); - goto got_one; - } - return NULL; -} - -void dlm_lowcomms_commit_buffer(void *mh) -{ - struct writequeue_entry *e = (struct writequeue_entry *)mh; - struct connection *con = e->con; - int users; - - spin_lock(&con->writequeue_lock); - users = --e->users; - if (users) - goto out; - e->len = e->end - e->offset; - kunmap(e->page); - spin_unlock(&con->writequeue_lock); - - if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) { - queue_work(send_workqueue, &con->swork); - } - return; - -out: - spin_unlock(&con->writequeue_lock); - return; -} - -static void free_entry(struct writequeue_entry *e) -{ - __free_page(e->page); - kfree(e); -} - -/* Send a message */ -static void send_to_sock(struct connection *con) -{ - int ret = 0; - ssize_t(*sendpage) (struct socket *, struct page *, int, size_t, int); - const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL; - struct writequeue_entry *e; - int len, offset; - - mutex_lock(&con->sock_mutex); - if (con->sock == NULL) - goto out_connect; - - sendpage = con->sock->ops->sendpage; - - spin_lock(&con->writequeue_lock); - for (;;) { - e = list_entry(con->writequeue.next, struct writequeue_entry, - list); - if ((struct list_head *) e == &con->writequeue) - break; - - len = e->len; - offset = e->offset; - BUG_ON(len == 0 && e->users == 0); - spin_unlock(&con->writequeue_lock); - kmap(e->page); - - ret = 0; - if (len) { - ret = sendpage(con->sock, e->page, offset, len, - msg_flags); - if (ret == -EAGAIN || ret == 0) - goto out; - if (ret <= 0) - goto send_error; - } - else { - /* Don't starve people filling buffers */ - cond_resched(); - } - - spin_lock(&con->writequeue_lock); - e->offset += ret; - e->len -= ret; - - if (e->len == 0 && e->users == 0) { - list_del(&e->list); - kunmap(e->page); - free_entry(e); - continue; - } - } - spin_unlock(&con->writequeue_lock); -out: - mutex_unlock(&con->sock_mutex); - return; - -send_error: - mutex_unlock(&con->sock_mutex); - close_connection(con, false); - lowcomms_connect_sock(con); - return; - -out_connect: - mutex_unlock(&con->sock_mutex); - connect_to_sock(con); - return; -} - -static void clean_one_writequeue(struct connection *con) -{ - struct list_head *list; - struct list_head *temp; - - spin_lock(&con->writequeue_lock); - list_for_each_safe(list, temp, &con->writequeue) { - struct writequeue_entry *e = - list_entry(list, struct writequeue_entry, list); - list_del(&e->list); - free_entry(e); - } - spin_unlock(&con->writequeue_lock); -} - -/* Called from recovery when it knows that a node has - left the cluster */ -int dlm_lowcomms_close(int nodeid) -{ - struct connection *con; - - if (!connections) - goto out; - - log_print("closing connection to node %d", nodeid); - con = nodeid2con(nodeid, 0); - if (con) { - clean_one_writequeue(con); - close_connection(con, true); - } - return 0; - -out: - return -1; -} - -/* Look for activity on active sockets */ -static void process_recv_sockets(struct work_struct *work) -{ - struct connection *con = container_of(work, struct connection, rwork); - int err; - - clear_bit(CF_READ_PENDING, &con->flags); - do { - err = con->rx_action(con); - } while (!err); -} - - -static void process_send_sockets(struct work_struct *work) -{ - struct connection *con = container_of(work, struct connection, swork); - - if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) { - connect_to_sock(con); - } - - clear_bit(CF_WRITE_PENDING, &con->flags); - send_to_sock(con); -} - - -/* Discard all entries on the write queues */ -static void clean_writequeues(void) -{ - int nodeid; - - for (nodeid = 1; nodeid < conn_array_size; nodeid++) { - struct connection *con = nodeid2con(nodeid, 0); - - if (con) - clean_one_writequeue(con); - } -} - -static void work_stop(void) -{ - destroy_workqueue(recv_workqueue); - destroy_workqueue(send_workqueue); -} - -static int work_start(void) -{ - int error; - recv_workqueue = create_workqueue("dlm_recv"); - error = IS_ERR(recv_workqueue); - if (error) { - log_print("can't start dlm_recv %d", error); - return error; - } - - send_workqueue = create_singlethread_workqueue("dlm_send"); - error = IS_ERR(send_workqueue); - if (error) { - log_print("can't start dlm_send %d", error); - destroy_workqueue(recv_workqueue); - return error; - } - - return 0; -} - -void dlm_lowcomms_stop(void) -{ - int i; - - /* Set all the flags to prevent any - socket activity. - */ - for (i = 0; i < conn_array_size; i++) { - if (connections[i]) - connections[i]->flags |= 0xFF; - } - - work_stop(); - clean_writequeues(); - - for (i = 0; i < conn_array_size; i++) { - if (connections[i]) { - close_connection(connections[i], true); - if (connections[i]->othercon) - kmem_cache_free(con_cache, connections[i]->othercon); - kmem_cache_free(con_cache, connections[i]); - } - } - - kfree(connections); - connections = NULL; - - kmem_cache_destroy(con_cache); -} - -/* This is quite likely to sleep... */ -int dlm_lowcomms_start(void) -{ - int error = 0; - - error = -ENOMEM; - connections = kzalloc(sizeof(struct connection *) * - NODE_INCREMENT, GFP_KERNEL); - if (!connections) - goto out; - - conn_array_size = NODE_INCREMENT; - - if (dlm_our_addr(&dlm_local_addr, 0)) { - log_print("no local IP address has been set"); - goto fail_free_conn; - } - if (!dlm_our_addr(&dlm_local_addr, 1)) { - log_print("This dlm comms module does not support multi-homed clustering"); - goto fail_free_conn; - } - - con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection), - __alignof__(struct connection), 0, - NULL, NULL); - if (!con_cache) - goto fail_free_conn; - - - /* Start listening */ - error = listen_for_all(); - if (error) - goto fail_unlisten; - - error = work_start(); - if (error) - goto fail_unlisten; - - return 0; - -fail_unlisten: - close_connection(connections[0], false); - kmem_cache_free(con_cache, connections[0]); - kmem_cache_destroy(con_cache); - -fail_free_conn: - kfree(connections); - -out: - return error; -} - -/* - * Overrides for Emacs so that we follow Linus's tabbing style. - * Emacs will notice this stuff at the end of the file and automatically - * adjust the settings for this buffer only. This must remain at the end - * of the file. - * --------------------------------------------------------------------------- - * Local variables: - * c-file-style: "linux" - * End: - */ |