*******************************************************************************
**
** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
-** Copyright (C) 2004-2007 Red Hat, Inc. All rights reserved.
+** Copyright (C) 2004-2009 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
*
* 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
+ * be expanded for the cluster infrastructure then that is its
* responsibility. It is this layer's
* responsibility to resolve these into IP address or
* whatever it needs for inter-node communication.
* of high load. Also, this way, the sending thread can collect together
* messages bound for one node and send them in one block.
*
- * lowcomms will choose to use wither TCP or SCTP as its transport layer
+ * lowcomms will choose to use either TCP or SCTP as its transport layer
* depending on the configuration variable 'protocol'. This should be set
- * to 0 (default) for TCP or 1 for SCTP. It shouldbe configured using a
+ * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
* cluster-wide mechanism as it must be the same on all nodes of the cluster
* for the DLM to function.
*
#include <net/sock.h>
#include <net/tcp.h>
#include <linux/pagemap.h>
-#include <linux/idr.h>
#include <linux/file.h>
+#include <linux/mutex.h>
#include <linux/sctp.h>
#include <net/sctp/user.h>
+#include <net/ipv6.h>
#include "dlm_internal.h"
#include "lowcomms.h"
#include "config.h"
#define NEEDED_RMEM (4*1024*1024)
+#define CONN_HASH_SIZE 32
struct cbuf {
unsigned int base;
int retries;
#define MAX_CONNECT_RETRIES 3
int sctp_assoc;
+ struct hlist_node list;
struct connection *othercon;
struct work_struct rwork; /* Receive workqueue */
struct work_struct swork; /* Send workqueue */
static struct workqueue_struct *recv_workqueue;
static struct workqueue_struct *send_workqueue;
-static DEFINE_IDR(connections_idr);
-static DECLARE_MUTEX(connections_lock);
-static int max_nodeid;
+static struct hlist_head connection_hash[CONN_HASH_SIZE];
+static DEFINE_MUTEX(connections_lock);
static struct kmem_cache *con_cache;
static void process_recv_sockets(struct work_struct *work);
static void process_send_sockets(struct work_struct *work);
+
+/* This is deliberately very simple because most clusters have simple
+ sequential nodeids, so we should be able to go straight to a connection
+ struct in the array */
+static inline int nodeid_hash(int nodeid)
+{
+ return nodeid & (CONN_HASH_SIZE-1);
+}
+
+static struct connection *__find_con(int nodeid)
+{
+ int r;
+ struct hlist_node *h;
+ struct connection *con;
+
+ r = nodeid_hash(nodeid);
+
+ hlist_for_each_entry(con, h, &connection_hash[r], list) {
+ if (con->nodeid == nodeid)
+ return con;
+ }
+ return NULL;
+}
+
/*
* If 'allocation' is zero then we don't attempt to create a new
* connection structure for this node.
{
struct connection *con = NULL;
int r;
- int n;
- con = idr_find(&connections_idr, nodeid);
+ con = __find_con(nodeid);
if (con || !alloc)
return con;
- r = idr_pre_get(&connections_idr, alloc);
- if (!r)
- return NULL;
-
con = kmem_cache_zalloc(con_cache, alloc);
if (!con)
return NULL;
- r = idr_get_new_above(&connections_idr, con, nodeid, &n);
- if (r) {
- kmem_cache_free(con_cache, con);
- return NULL;
- }
-
- if (n != nodeid) {
- idr_remove(&connections_idr, n);
- kmem_cache_free(con_cache, con);
- return NULL;
- }
+ r = nodeid_hash(nodeid);
+ hlist_add_head(&con->list, &connection_hash[r]);
con->nodeid = nodeid;
mutex_init(&con->sock_mutex);
/* Setup action pointers for child sockets */
if (con->nodeid) {
- struct connection *zerocon = idr_find(&connections_idr, 0);
+ struct connection *zerocon = __find_con(0);
con->connect_action = zerocon->connect_action;
if (!con->rx_action)
con->rx_action = zerocon->rx_action;
}
- if (nodeid > max_nodeid)
- max_nodeid = nodeid;
-
return con;
}
+/* Loop round all connections */
+static void foreach_conn(void (*conn_func)(struct connection *c))
+{
+ int i;
+ struct hlist_node *h, *n;
+ struct connection *con;
+
+ for (i = 0; i < CONN_HASH_SIZE; i++) {
+ hlist_for_each_entry_safe(con, h, n, &connection_hash[i], list){
+ conn_func(con);
+ }
+ }
+}
+
static struct connection *nodeid2con(int nodeid, gfp_t allocation)
{
struct connection *con;
- down(&connections_lock);
+ mutex_lock(&connections_lock);
con = __nodeid2con(nodeid, allocation);
- up(&connections_lock);
+ mutex_unlock(&connections_lock);
return con;
}
static struct connection *assoc2con(int assoc_id)
{
int i;
+ struct hlist_node *h;
struct connection *con;
- down(&connections_lock);
- for (i=0; i<=max_nodeid; i++) {
- con = __nodeid2con(i, 0);
- if (con && con->sctp_assoc == assoc_id) {
- up(&connections_lock);
- return con;
+ mutex_lock(&connections_lock);
+
+ for (i = 0 ; i < CONN_HASH_SIZE; i++) {
+ hlist_for_each_entry(con, h, &connection_hash[i], list) {
+ if (con && con->sctp_assoc == assoc_id) {
+ mutex_unlock(&connections_lock);
+ return con;
+ }
}
}
- up(&connections_lock);
+ mutex_unlock(&connections_lock);
return NULL;
}
} else {
struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &addr;
struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) retaddr;
- memcpy(&ret6->sin6_addr, &in6->sin6_addr,
- sizeof(in6->sin6_addr));
+ ipv6_addr_copy(&ret6->sin6_addr, &in6->sin6_addr);
}
return 0;
lowcomms_write_space(sk);
}
+int dlm_lowcomms_connect_node(int nodeid)
+{
+ struct connection *con;
+
+ if (nodeid == dlm_our_nodeid())
+ return 0;
+
+ con = nodeid2con(nodeid, GFP_NOFS);
+ if (!con)
+ return -ENOMEM;
+ lowcomms_connect_sock(con);
+ return 0;
+}
+
/* Make a socket active */
static int add_sock(struct socket *sock, struct connection *con)
{
con->sock->sk->sk_write_space = lowcomms_write_space;
con->sock->sk->sk_state_change = lowcomms_state_change;
con->sock->sk->sk_user_data = con;
+ con->sock->sk->sk_allocation = GFP_NOFS;
return 0;
}
log_print("send EOF to node failed: %d", ret);
}
+static void sctp_init_failed_foreach(struct connection *con)
+{
+ con->sctp_assoc = 0;
+ if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
+ if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
+ queue_work(send_workqueue, &con->swork);
+ }
+}
+
/* INIT failed but we don't know which node...
restart INIT on all pending nodes */
static void sctp_init_failed(void)
{
- int i;
- struct connection *con;
+ mutex_lock(&connections_lock);
- down(&connections_lock);
- for (i=1; i<=max_nodeid; i++) {
- con = __nodeid2con(i, 0);
- if (!con)
- continue;
- con->sctp_assoc = 0;
- if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
- if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {
- queue_work(send_workqueue, &con->swork);
- }
- }
- }
- up(&connections_lock);
+ foreach_conn(sctp_init_failed_foreach);
+
+ mutex_unlock(&connections_lock);
}
/* Something happened to an association */
return;
}
- new_con = nodeid2con(nodeid, GFP_KERNEL);
+ new_con = nodeid2con(nodeid, GFP_NOFS);
if (!new_con)
return;
* the same time and the connections cross on the wire.
* In this case we store the incoming one in "othercon"
*/
- newcon = nodeid2con(nodeid, GFP_KERNEL);
+ newcon = nodeid2con(nodeid, GFP_NOFS);
if (!newcon) {
result = -ENOMEM;
goto accept_err;
struct connection *othercon = newcon->othercon;
if (!othercon) {
- othercon = kmem_cache_zalloc(con_cache, GFP_KERNEL);
+ othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
if (!othercon) {
log_print("failed to allocate incoming socket");
mutex_unlock(&newcon->sock_mutex);
len = e->len;
offset = e->offset;
spin_unlock(&con->writequeue_lock);
- kmap(e->page);
/* Send the first block off the write queue */
iov[0].iov_base = page_address(e->page)+offset;
if (e->len == 0 && e->users == 0) {
list_del(&e->list);
- kunmap(e->page);
free_entry(e);
}
spin_unlock(&con->writequeue_lock);
static void tcp_connect_to_sock(struct connection *con)
{
int result = -EHOSTUNREACH;
- struct sockaddr_storage saddr;
+ struct sockaddr_storage saddr, src_addr;
int addr_len;
- struct socket *sock;
+ struct socket *sock = NULL;
if (con->nodeid == 0) {
log_print("attempt to connect sock 0 foiled");
goto out_err;
memset(&saddr, 0, sizeof(saddr));
- if (dlm_nodeid_to_addr(con->nodeid, &saddr))
+ if (dlm_nodeid_to_addr(con->nodeid, &saddr)) {
+ sock_release(sock);
goto out_err;
+ }
sock->sk->sk_user_data = con;
con->rx_action = receive_from_sock;
con->connect_action = tcp_connect_to_sock;
add_sock(sock, con);
+ /* Bind to our cluster-known address connecting to avoid
+ routing problems */
+ memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
+ make_sockaddr(&src_addr, 0, &addr_len);
+ result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
+ addr_len);
+ if (result < 0) {
+ log_print("could not bind for connect: %d", result);
+ /* This *may* not indicate a critical error */
+ }
+
make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
log_print("connecting to %d", con->nodeid);
if (con->sock) {
sock_release(con->sock);
con->sock = NULL;
+ } else if (sock) {
+ sock_release(sock);
}
/*
* 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 != -ENETDOWN && result != -EINVAL
&& result != -EPROTONOSUPPORT) {
lowcomms_connect_sock(con);
result = 0;
if (e) {
got_one:
- if (users == 0)
- kmap(e->page);
*ppc = page_address(e->page) + offset;
return e;
}
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)) {
offset = e->offset;
BUG_ON(len == 0 && e->users == 0);
spin_unlock(&con->writequeue_lock);
- kmap(e->page);
ret = 0;
if (len) {
if (e->len == 0 && e->users == 0) {
list_del(&e->list);
- kunmap(e->page);
free_entry(e);
continue;
}
static void clean_one_writequeue(struct connection *con)
{
- struct list_head *list;
- struct list_head *temp;
+ struct writequeue_entry *e, *safe;
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_for_each_entry_safe(e, safe, &con->writequeue, list) {
list_del(&e->list);
free_entry(e);
}
/* Discard all entries on the write queues */
static void clean_writequeues(void)
{
- int nodeid;
-
- for (nodeid = 1; nodeid <= max_nodeid; nodeid++) {
- struct connection *con = __nodeid2con(nodeid, 0);
-
- if (con)
- clean_one_writequeue(con);
- }
+ foreach_conn(clean_one_writequeue);
}
static void work_stop(void)
return 0;
}
-void dlm_lowcomms_stop(void)
+static void stop_conn(struct connection *con)
{
- int i;
- struct connection *con;
+ con->flags |= 0x0F;
+ if (con->sock && con->sock->sk)
+ con->sock->sk->sk_user_data = NULL;
+}
+static void free_conn(struct connection *con)
+{
+ close_connection(con, true);
+ if (con->othercon)
+ kmem_cache_free(con_cache, con->othercon);
+ hlist_del(&con->list);
+ kmem_cache_free(con_cache, con);
+}
+
+void dlm_lowcomms_stop(void)
+{
/* Set all the flags to prevent any
socket activity.
*/
- down(&connections_lock);
- for (i = 0; i <= max_nodeid; i++) {
- con = __nodeid2con(i, 0);
- if (con) {
- con->flags |= 0x0F;
- if (con->sock)
- con->sock->sk->sk_user_data = NULL;
- }
- }
- up(&connections_lock);
+ mutex_lock(&connections_lock);
+ foreach_conn(stop_conn);
+ mutex_unlock(&connections_lock);
work_stop();
- down(&connections_lock);
+ mutex_lock(&connections_lock);
clean_writequeues();
- for (i = 0; i <= max_nodeid; i++) {
- con = __nodeid2con(i, 0);
- if (con) {
- close_connection(con, true);
- kmem_cache_free(con_cache, con);
- }
- }
- max_nodeid = 0;
- up(&connections_lock);
+ foreach_conn(free_conn);
+
+ mutex_unlock(&connections_lock);
kmem_cache_destroy(con_cache);
- idr_init(&connections_idr);
}
int dlm_lowcomms_start(void)
{
int error = -EINVAL;
struct connection *con;
+ int i;
+
+ for (i = 0; i < CONN_HASH_SIZE; i++)
+ INIT_HLIST_HEAD(&connection_hash[i]);
init_local();
if (!dlm_local_count) {
git://ftp.safe.ca / safe / jmp / linux-2.6 / blobdiff