cachesyncd/selector.c

#include <stdlib.h>
#include <unistd.h>
#include <string.h>

#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>

#include "list.h"
#include "logging.h"
#include "multicast.h"
#include "unixsock.h"

#define BUF_SIZE 256

struct fd_entry {
	struct list_head list;
	int fd;
};

static LIST_HEAD(fd_list);

void selector(int usock, int msock)
{
	fd_set fdsel;

	char *buf = malloc(BUF_SIZE);
	if (buf == NULL) {
		log_print(LOG_ERROR, "selector: out of memory");
		return;
	}

	while (1) {
		struct timeval tv;
		struct fd_entry *entry, *entry_safe;

		FD_ZERO(&fdsel);
		FD_SET(usock, &fdsel);
		FD_SET(msock, &fdsel);

		list_for_each_entry(entry, &fd_list, list) {
			FD_SET(entry->fd, &fdsel);
		}

		tv.tv_sec = 60;
		tv.tv_usec = 0;

		int ret = select(FD_SETSIZE, &fdsel, NULL, NULL, &tv);
		if (ret < 0) {
			log_print(LOG_ERROR, "selector: select()");
			continue;

		} else if (ret == 0) {
			mcast_send(msock, "KEEPALIVE", 10);
		}

		if (FD_ISSET(msock, &fdsel)) {
			int len = read(msock, buf, BUF_SIZE);
			if (len <= 0) {
				log_print(LOG_ERROR, "selector: multicast sock closed?");

			} else if (!strncmp(buf, "KEEPALIVE", 10)) {
				// nothing

			} else if (!strncmp(buf, "DELETE ", 7)) {
				log_print(LOG_DEBUG, "delete '%s'", buf +7);
//				delete_file(buf +7);

			} else {
				log_print(LOG_DEBUG, "recv unknown cmd via multicast: '%s'", buf);
			}
		}

		if (FD_ISSET(usock, &fdsel)) {
			int usock_con;

			usock_con = accept(usock, NULL, NULL);
			if (usock_con != -1) {
				entry = malloc(sizeof(struct fd_entry));
				if (entry) {
					entry->fd = usock_con;
					list_add_tail(&entry->list, &fd_list);
				} else {
					log_print(LOG_ERROR, "selector: out of memory");
					close(usock_con);
				}

			} else {
				log_print(LOG_ERROR, "selector: accept()");
			}
		}

		list_for_each_entry_safe(entry, entry_safe, &fd_list, list) {
			if (FD_ISSET(entry->fd, &fdsel)) {
				int len = read(entry->fd, buf, BUF_SIZE);
				if (len <= 0) {
					list_del(&entry->list);
					close(entry->fd);
					free(entry);

				} else {
					mcast_send(msock, buf, len);
				}
			}
		}
	}
	msock_close(msock);
	sock_close(usock);
}