ip_vs_sync.c

linux-2.4.29操作系统的源码

	return sock->ops->bind(sock, (struct sockaddr*)&sin, sizeof(sin));
}

/*
 *      Set up sending multicast socket over UDP
 */
static struct socket * make_send_sock(void)
{
	struct socket *sock;

	/* First create a socket */
	if (sock_create(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock) < 0) {
		IP_VS_ERR("Error during creation of socket; terminating\n");
		return NULL;
	}

	if (set_mcast_if(sock->sk, ip_vs_mcast_master_ifn) < 0) {
		IP_VS_ERR("Error setting outbound mcast interface\n");
		goto error;
	}

	set_mcast_loop(sock->sk, 0);
	set_mcast_ttl(sock->sk, 1);

	if (bind_mcastif_addr(sock, ip_vs_mcast_master_ifn) < 0) {
		IP_VS_ERR("Error binding address of the mcast interface\n");
		goto error;
	}

	if (sock->ops->connect(sock,
			       (struct sockaddr*)&mcast_addr,
			       sizeof(struct sockaddr), 0) < 0) {
		IP_VS_ERR("Error connecting to the multicast addr\n");
		goto error;
	}

	return sock;

  error:
	sock_release(sock);
	return NULL;
}


/*
 *      Set up receiving multicast socket over UDP
 */
static struct socket * make_receive_sock(void)
{
	struct socket *sock;

	/* First create a socket */
	if (sock_create(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock) < 0) {
		IP_VS_ERR("Error during creation of socket; terminating\n");
		return NULL;
	}

	/* it is equivalent to the REUSEADDR option in user-space */
	sock->sk->reuse = 1;

	if (sock->ops->bind(sock,
			    (struct sockaddr*)&mcast_addr,
			    sizeof(struct sockaddr)) < 0) {
		IP_VS_ERR("Error binding to the multicast addr\n");
		goto error;
	}

	/* join the multicast group */
	if (join_mcast_group(sock->sk,
			     (struct in_addr*)&mcast_addr.sin_addr,
			     ip_vs_mcast_backup_ifn) < 0) {
		IP_VS_ERR("Error joining to the multicast group\n");
		goto error;
	}

	return sock;

  error:
	sock_release(sock);
	return NULL;
}


static int
ip_vs_send_async(struct socket *sock, const char *buffer, const size_t length)
{
	struct msghdr	msg;
	mm_segment_t	oldfs;
	struct iovec	iov;
	int		len;

	EnterFunction(7);
	iov.iov_base     = (void *)buffer;
	iov.iov_len      = length;
	msg.msg_name     = 0;
	msg.msg_namelen  = 0;
	msg.msg_iov	 = &iov;
	msg.msg_iovlen   = 1;
	msg.msg_control  = NULL;
	msg.msg_controllen = 0;
	msg.msg_flags    = MSG_DONTWAIT|MSG_NOSIGNAL;

	oldfs = get_fs(); set_fs(KERNEL_DS);
	len = sock_sendmsg(sock, &msg, (size_t)(length));
	set_fs(oldfs);

	LeaveFunction(7);
	return len;
}

static void
ip_vs_send_sync_msg(struct socket *sock, struct ip_vs_sync_mesg *msg)
{
	int msize;

	msize = msg->size;

	/* Put size in network byte order */
	msg->size = htons(msg->size);

	if (ip_vs_send_async(sock, (char *)msg, msize) != msize)
		IP_VS_ERR("ip_vs_send_async error\n");
}

static int
ip_vs_receive(struct socket *sock, char *buffer, const size_t buflen)
{
	struct msghdr		msg;
	struct iovec		iov;
	int			len;
	mm_segment_t		oldfs;

	EnterFunction(7);

	/* Receive a packet */
	iov.iov_base     = buffer;
	iov.iov_len      = (size_t)buflen;
	msg.msg_name     = 0;
	msg.msg_namelen  = 0;
	msg.msg_iov	 = &iov;
	msg.msg_iovlen   = 1;
	msg.msg_control  = NULL;
	msg.msg_controllen = 0;
	msg.msg_flags    = 0;

	oldfs = get_fs(); set_fs(KERNEL_DS);
	len = sock_recvmsg(sock, &msg, buflen, 0);
	set_fs(oldfs);

	if (len < 0)
		return -1;

	LeaveFunction(7);
	return len;
}


static DECLARE_WAIT_QUEUE_HEAD(sync_wait);
static pid_t sync_master_pid = 0;
static pid_t sync_backup_pid = 0;

static DECLARE_WAIT_QUEUE_HEAD(stop_sync_wait);
static int stop_master_sync = 0;
static int stop_backup_sync = 0;

static void sync_master_loop(void)
{
	struct socket *sock;
	struct ip_vs_sync_buff *sb;

	/* create the sending multicast socket */
	sock = make_send_sock();
	if (!sock)
		return;

	IP_VS_INFO("sync thread started: state = MASTER, mcast_ifn = %s, "
		   "syncid = %d\n",
		   ip_vs_mcast_master_ifn, ip_vs_master_syncid);

	for (;;) {
		while ((sb=sb_dequeue())) {
			ip_vs_send_sync_msg(sock, sb->mesg);
			ip_vs_sync_buff_release(sb);
		}

		/* check if entries stay in curr_sb for 2 seconds */
		if ((sb = get_curr_sync_buff(2*HZ))) {
			ip_vs_send_sync_msg(sock, sb->mesg);
			ip_vs_sync_buff_release(sb);
		}

		if (stop_master_sync)
			break;

		__set_current_state(TASK_INTERRUPTIBLE);
		schedule_timeout(HZ);
		__set_current_state(TASK_RUNNING);
	}

	/* clean up the sync_buff queue */
	while ((sb=sb_dequeue())) {
		ip_vs_sync_buff_release(sb);
	}

	/* clean up the current sync_buff */
	if ((sb = get_curr_sync_buff(0))) {
		ip_vs_sync_buff_release(sb);
	}

	/* release the sending multicast socket */
	sock_release(sock);
}


static void sync_backup_loop(void)
{
	struct socket *sock;
	char *buf;
	int len;

	if (!(buf=kmalloc(SYNC_MESG_MAX_SIZE, GFP_ATOMIC))) {
		IP_VS_ERR("sync_backup_loop: kmalloc error\n");
		return;
	}

	/* create the receiving multicast socket */
	sock = make_receive_sock();
	if (!sock)
		goto out;

	IP_VS_INFO("sync thread started: state = BACKUP, mcast_ifn = %s, "
		   "syncid = %d\n",
		   ip_vs_mcast_backup_ifn, ip_vs_backup_syncid);

	for (;;) {
		/* do you have data now? */
		while (!skb_queue_empty(&(sock->sk->receive_queue))) {
			if ((len=ip_vs_receive(sock, buf,
					       SYNC_MESG_MAX_SIZE))<=0) {
				IP_VS_ERR("receiving message error\n");
				break;
			}
			/* disable bottom half, because it accessed the data
			   shared by softirq while getting/creating conns */
			local_bh_disable();
			ip_vs_process_message(buf, len);
			local_bh_enable();
		}

		if (stop_backup_sync)
			break;

		__set_current_state(TASK_INTERRUPTIBLE);
		schedule_timeout(HZ);
		__set_current_state(TASK_RUNNING);
	}

	/* release the sending multicast socket */
	sock_release(sock);

  out:
	kfree(buf);
}

static void sync_pid_set(int sync_state, pid_t sync_pid)
{
	if (sync_state == IP_VS_STATE_MASTER)
		sync_master_pid = sync_pid;
	else if (sync_state == IP_VS_STATE_BACKUP)
		sync_backup_pid = sync_pid;
}

static void sync_stop_set(int sync_state, int set)
{
	if (sync_state == IP_VS_STATE_MASTER)
		stop_master_sync = set;
	else if (sync_state == IP_VS_STATE_BACKUP)
		stop_backup_sync = set;
	else {
		stop_master_sync = set;
		stop_backup_sync = set;
	}
}

static int sync_thread(void *startup)
{
	DECLARE_WAITQUEUE(wait, current);
	mm_segment_t oldmm;
	int state = IP_VS_STATE_NONE;

	MOD_INC_USE_COUNT;
	daemonize();

	oldmm = get_fs();
	set_fs(KERNEL_DS);

	if (ip_vs_sync_state & IP_VS_STATE_MASTER && !sync_master_pid) {
		state = IP_VS_STATE_MASTER;
		sprintf(current->comm, "ipvs_syncmaster");
	} else if (ip_vs_sync_state & IP_VS_STATE_BACKUP) {
		state = IP_VS_STATE_BACKUP;
		sprintf(current->comm, "ipvs_syncbackup");
	} else IP_VS_BUG();

	/* Block all signals */
	spin_lock_irq(&current->sigmask_lock);
	siginitsetinv(&current->blocked, 0);
	recalc_sigpending(current);
	spin_unlock_irq(&current->sigmask_lock);

	/* set up multicast address */
	mcast_addr.sin_family = AF_INET;
	mcast_addr.sin_port = htons(IP_VS_SYNC_PORT);
	mcast_addr.sin_addr.s_addr = htonl(IP_VS_SYNC_GROUP);

	add_wait_queue(&sync_wait, &wait);

	sync_pid_set(state, current->pid);
	complete((struct completion *)startup);

	/* processing master/backup loop here */
	if (state == IP_VS_STATE_MASTER)
		sync_master_loop();
	else if (state == IP_VS_STATE_BACKUP)
		sync_backup_loop();
	else IP_VS_BUG();

	remove_wait_queue(&sync_wait, &wait);

	/* thread exits */
	sync_pid_set(state, 0);
	IP_VS_INFO("sync thread stopped!\n");

	set_fs(oldmm);
	MOD_DEC_USE_COUNT;

	sync_stop_set(state, 0);
	wake_up(&stop_sync_wait);

	return 0;
}


static int fork_sync_thread(void *startup)
{
	pid_t pid;

	/* fork the sync thread here, then the parent process of the
	   sync thread is the init process after this thread exits. */
  repeat:
	if ((pid = kernel_thread(sync_thread, startup, 0)) < 0) {
		IP_VS_ERR("could not create sync_thread due to %d... "
			  "retrying.\n", pid);
		current->state = TASK_UNINTERRUPTIBLE;
		schedule_timeout(HZ);
		goto repeat;
	}

	return 0;
}


int start_sync_thread(int state, char *mcast_ifn, __u8 syncid)
{
	DECLARE_COMPLETION(startup);
	pid_t pid;

	if ((state == IP_VS_STATE_MASTER && sync_master_pid) ||
	    (state == IP_VS_STATE_BACKUP && sync_backup_pid))
		return -EEXIST;

	IP_VS_DBG(7, "%s: pid %d\n", __FUNCTION__, current->pid);
	IP_VS_DBG(7, "Each ip_vs_sync_conn entry need %d bytes\n",
		  sizeof(struct ip_vs_sync_conn));

	ip_vs_sync_state |= state;
	if (state == IP_VS_STATE_MASTER) {
		strcpy(ip_vs_mcast_master_ifn, mcast_ifn);
		ip_vs_master_syncid = syncid;
	} else {
		strcpy(ip_vs_mcast_backup_ifn, mcast_ifn);
		ip_vs_backup_syncid = syncid;
	}

  repeat:
	if ((pid = kernel_thread(fork_sync_thread, &startup, 0)) < 0) {
		IP_VS_ERR("could not create fork_sync_thread due to %d... "
			  "retrying.\n", pid);
		current->state = TASK_UNINTERRUPTIBLE;
		schedule_timeout(HZ);
		goto repeat;
	}

	wait_for_completion(&startup);

	return 0;
}


int stop_sync_thread(int state)
{
	DECLARE_WAITQUEUE(wait, current);

	if ((state == IP_VS_STATE_MASTER && !sync_master_pid) ||
	    (state == IP_VS_STATE_BACKUP && !sync_backup_pid))
		return -ESRCH;

	IP_VS_DBG(7, "%s: pid %d\n", __FUNCTION__, current->pid);
	IP_VS_INFO("stopping sync thread %d ...\n",
		   (state == IP_VS_STATE_MASTER) ? sync_master_pid : sync_backup_pid);

	__set_current_state(TASK_UNINTERRUPTIBLE);
	add_wait_queue(&stop_sync_wait, &wait);
	sync_stop_set(state, 1);
	ip_vs_sync_state -= state;
	wake_up(&sync_wait);
	schedule();
	__set_current_state(TASK_RUNNING);
	remove_wait_queue(&stop_sync_wait, &wait);

	/* Note: no need to reap the sync thread, because its parent
	   process is the init process */

	if ((state == IP_VS_STATE_MASTER && stop_master_sync) ||
	    (state == IP_VS_STATE_BACKUP && stop_backup_sync))
		IP_VS_BUG();

	return 0;
}