path: root/fs/dlm/lowcomms-tcp.c

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
**  Copyright (C) 2004-2006 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 rw_semaphore sock_sem; /* Stop connect races */
	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;
	atomic_t waiting_requests;
#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 = no