#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>
static struct workqueue_struct *send_workqueue;
static DEFINE_IDR(connections_idr);
-static DECLARE_MUTEX(connections_lock);
+static DEFINE_MUTEX(connections_lock);
static int max_nodeid;
static struct kmem_cache *con_cache;
{
struct connection *con;
- down(&connections_lock);
+ mutex_lock(&connections_lock);
con = __nodeid2con(nodeid, allocation);
- up(&connections_lock);
+ mutex_unlock(&connections_lock);
return con;
}
int i;
struct connection *con;
- down(&connections_lock);
- for (i=0; i<max_nodeid; i++) {
+ mutex_lock(&connections_lock);
+ for (i=0; i<=max_nodeid; i++) {
con = __nodeid2con(i, 0);
if (con && con->sctp_assoc == assoc_id) {
- up(&connections_lock);
+ mutex_unlock(&connections_lock);
return con;
}
}
- up(&connections_lock);
+ mutex_unlock(&connections_lock);
return NULL;
}
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))
+ if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
queue_work(recv_workqueue, &con->rwork);
}
{
struct connection *con = sock2con(sk);
- if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
+ if (con && !test_and_set_bit(CF_WRITE_PENDING, &con->flags))
queue_work(send_workqueue, &con->swork);
}
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;
}
in6_addr->sin6_port = cpu_to_be16(port);
*addr_len = sizeof(struct sockaddr_in6);
}
+ memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
}
/* Close a remote connection and tidy up */
__free_page(con->rx_page);
con->rx_page = NULL;
}
+
con->retries = 0;
mutex_unlock(&con->sock_mutex);
}
int i;
struct connection *con;
- down(&connections_lock);
+ mutex_lock(&connections_lock);
for (i=1; i<=max_nodeid; i++) {
con = __nodeid2con(i, 0);
if (!con)
}
}
}
- up(&connections_lock);
+ mutex_unlock(&connections_lock);
}
/* Something happened to an association */
-static void process_sctp_notification(struct connection *con, struct msghdr *msg, char *buf)
+static void process_sctp_notification(struct connection *con,
+ struct msghdr *msg, char *buf)
{
union sctp_notification *sn = (union sctp_notification *)buf;
*/
if ((int)sn->sn_assoc_change.sac_assoc_id <= 0) {
log_print("COMM_UP for invalid assoc ID %d",
- (int)sn->sn_assoc_change.sac_assoc_id);
+ (int)sn->sn_assoc_change.sac_assoc_id);
sctp_init_failed();
return;
}
/* Peel off a new sock */
parg.associd = sn->sn_assoc_change.sac_assoc_id;
- ret = kernel_getsockopt(con->sock, IPPROTO_SCTP, SCTP_SOCKOPT_PEELOFF,
+ ret = kernel_getsockopt(con->sock, IPPROTO_SCTP,
+ SCTP_SOCKOPT_PEELOFF,
(void *)&parg, &parglen);
- if (ret < 0) {
- log_print("Can't peel off a socket for connection %d to node %d: err=%d\n",
+ if (ret) {
+ log_print("Can't peel off a socket for "
+ "connection %d to node %d: err=%d\n",
parg.associd, nodeid, ret);
- return;
}
-
file = fget(parg.sd);
new_con->sock = SOCKET_I(file->f_dentry->d_inode);
add_sock(new_con->sock, new_con);
put_unused_fd(parg.sd);
log_print("got new/restarted association %d nodeid %d",
- (int)sn->sn_assoc_change.sac_assoc_id, nodeid);
+ (int)sn->sn_assoc_change.sac_assoc_id, nodeid);
/* Send any pending writes */
clear_bit(CF_CONNECT_PENDING, &new_con->flags);
/* Process SCTP notifications */
if (msg.msg_flags & MSG_NOTIFICATION) {
- BUG_ON(con->nodeid != 0);
msg.msg_control = incmsg;
msg.msg_controllen = sizeof(incmsg);
process_sctp_notification(con, &msg,
- page_address(con->rx_page) + con->cb.base);
+ page_address(con->rx_page) + con->cb.base);
mutex_unlock(&con->sock_mutex);
return 0;
}
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);
+ log_print("lowcomms: addr=%p, base=%u, len=%u, "
+ "iov_len=%u, iov_base[0]=%p, read=%d",
+ page_address(con->rx_page), con->cb.base, con->cb.len,
+ len, iov[0].iov_base, r);
}
if (ret < 0)
goto out_close;
out_close:
mutex_unlock(&con->sock_mutex);
- if (ret != -EAGAIN && !test_bit(CF_IS_OTHERCON, &con->flags)) {
+ if (ret != -EAGAIN) {
close_connection(con, false);
/* Reconnect when there is something to send */
}
/* 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");
+ log_print("connect from non cluster node");
sock_release(newsock);
mutex_unlock(&con->sock_mutex);
return -1;
if (!othercon) {
othercon = kmem_cache_zalloc(con_cache, GFP_KERNEL);
if (!othercon) {
- printk("dlm: failed to allocate incoming socket\n");
+ log_print("failed to allocate incoming socket");
mutex_unlock(&newcon->sock_mutex);
result = -ENOMEM;
goto accept_err;
INIT_WORK(&othercon->swork, process_send_sockets);
INIT_WORK(&othercon->rwork, process_recv_sockets);
set_bit(CF_IS_OTHERCON, &othercon->flags);
+ }
+ if (!othercon->sock) {
newcon->othercon = othercon;
+ othercon->sock = newsock;
+ newsock->sk->sk_user_data = othercon;
+ add_sock(newsock, othercon);
+ addcon = othercon;
+ }
+ else {
+ printk("Extra connection from node %d attempted\n", nodeid);
+ result = -EAGAIN;
+ mutex_unlock(&newcon->sock_mutex);
+ goto accept_err;
}
- othercon->sock = newsock;
- newsock->sk->sk_user_data = othercon;
- add_sock(newsock, othercon);
- addcon = othercon;
}
else {
newsock->sk->sk_user_data = newcon;
sock_release(newsock);
if (result != -EAGAIN)
- printk("dlm: error accepting connection from node: %d\n", result);
+ log_print("error accepting connection from node: %d", result);
return result;
}
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;
ret = kernel_sendmsg(base_con->sock, &outmessage, iov, 1, len);
if (ret < 0) {
- log_print("Send first packet to node %d failed: %d", con->nodeid, ret);
+ log_print("Send first packet to node %d failed: %d",
+ con->nodeid, ret);
/* Try again later */
clear_bit(CF_CONNECT_PENDING, &con->flags);
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;
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);
* 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;
addr_len = sizeof(struct sockaddr_in6);
/* Create a socket to communicate with */
- result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM, IPPROTO_TCP, &sock);
+ result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
+ IPPROTO_TCP, &sock);
if (result < 0) {
- printk("dlm: Can't create listening comms socket\n");
+ log_print("Can't create listening comms socket");
goto create_out;
}
(char *)&one, sizeof(one));
if (result < 0) {
- printk("dlm: Failed to set SO_REUSEADDR on socket: result=%d\n",
- result);
+ log_print("Failed to set SO_REUSEADDR on socket: %d", result);
}
sock->sk->sk_user_data = con;
con->rx_action = tcp_accept_from_sock;
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);
+ log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
sock_release(sock);
sock = NULL;
con->sock = NULL;
result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
(char *)&one, sizeof(one));
if (result < 0) {
- printk("dlm: Set keepalive failed: %d\n", result);
+ log_print("Set keepalive failed: %d", result);
}
result = sock->ops->listen(sock, 5);
if (result < 0) {
- printk("dlm: Can't listen on port %d\n", dlm_config.ci_tcp_port);
+ log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
sock_release(sock);
sock = NULL;
goto create_out;
struct sockaddr_storage sas, *addr;
int i;
+ dlm_local_count = 0;
for (i = 0; i < DLM_MAX_ADDR_COUNT - 1; i++) {
if (dlm_our_addr(&sas, i))
break;
}
}
-/* Bind to an IP address. SCTP allows multiple address so it can do multi-homing */
-static int add_sctp_bind_addr(struct connection *sctp_con, struct sockaddr_storage *addr, int addr_len, int num)
+/* Bind to an IP address. SCTP allows multiple address so it can do
+ multi-homing */
+static int add_sctp_bind_addr(struct connection *sctp_con,
+ struct sockaddr_storage *addr,
+ int addr_len, int num)
{
int result = 0;
subscribe.sctp_shutdown_event = 1;
subscribe.sctp_partial_delivery_event = 1;
- result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUF,
+ result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
(char *)&bufsize, sizeof(bufsize));
if (result)
- log_print("Error increasing buffer space on socket: %d", result);
+ log_print("Error increasing buffer space on socket %d", result);
result = kernel_setsockopt(sock, SOL_SCTP, SCTP_EVENTS,
- (char *)&subscribe, sizeof(subscribe));
+ (char *)&subscribe, sizeof(subscribe));
if (result < 0) {
log_print("Failed to set SCTP_EVENTS on socket: result=%d",
result);
/* We don't support multi-homed hosts */
if (dlm_local_addr[1] != NULL) {
- log_print("TCP protocol can't handle multi-homed hosts, try SCTP");
+ log_print("TCP protocol can't handle multi-homed hosts, "
+ "try SCTP");
return -EINVAL;
}
log_print("Using TCP for communications");
- set_bit(CF_IS_OTHERCON, &con->flags);
-
sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
if (sock) {
add_sock(sock, con);
return entry;
}
-void *dlm_lowcomms_get_buffer(int nodeid, int len,
- gfp_t allocation, char **ppc)
+void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
{
struct connection *con;
struct writequeue_entry *e;
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) {
ret = sendpage(con->sock, e->page, offset, len,
msg_flags);
- if (ret == -EAGAIN || ret == 0)
+ if (ret == -EAGAIN || ret == 0) {
+ cond_resched();
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;
if (e->len == 0 && e->users == 0) {
list_del(&e->list);
- kunmap(e->page);
free_entry(e);
continue;
}
{
int nodeid;
- for (nodeid = 1; nodeid < max_nodeid; nodeid++) {
- struct connection *con = nodeid2con(nodeid, 0);
+ for (nodeid = 1; nodeid <= max_nodeid; nodeid++) {
+ struct connection *con = __nodeid2con(nodeid, 0);
if (con)
clean_one_writequeue(con);
/* Set all the flags to prevent any
socket activity.
*/
- down(&connections_lock);
- for (i = 0; i < max_nodeid; i++) {
+ mutex_lock(&connections_lock);
+ for (i = 0; i <= max_nodeid; i++) {
con = __nodeid2con(i, 0);
- if (con)
- con->flags |= 0xFF;
+ if (con) {
+ con->flags |= 0x0F;
+ if (con->sock)
+ con->sock->sk->sk_user_data = NULL;
+ }
}
- up(&connections_lock);
+ mutex_unlock(&connections_lock);
work_stop();
- down(&connections_lock);
+ mutex_lock(&connections_lock);
clean_writequeues();
- for (i = 0; i < max_nodeid; i++) {
+ for (i = 0; i <= max_nodeid; i++) {
con = __nodeid2con(i, 0);
if (con) {
close_connection(con, true);
kmem_cache_free(con_cache, con);
}
}
- up(&connections_lock);
+ max_nodeid = 0;
+ mutex_unlock(&connections_lock);
kmem_cache_destroy(con_cache);
+ idr_init(&connections_idr);
}
int dlm_lowcomms_start(void)
init_local();
if (!dlm_local_count) {
+ error = -ENOTCONN;
log_print("no local IP address has been set");
goto out;
}
error = -ENOMEM;
con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
__alignof__(struct connection), 0,
- NULL, NULL);
+ NULL);
if (!con_cache)
goto out;
- /* Set some sysctl minima */
- if (sysctl_rmem_max < NEEDED_RMEM)
- sysctl_rmem_max = NEEDED_RMEM;
-
/* Start listening */
if (dlm_config.ci_protocol == 0)
error = tcp_listen_for_all();
git://ftp.safe.ca / safe / jmp / linux-2.6 / blobdiff