protocol_s.c

使用具有增量学习的监控式学习方法。包括几个不同的分类算法。

/* Copyright (C) 2002  Mikael Ylikoski
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, Version 2 as
 * published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
 */
/**
 * @file
 * Classification daemon protocol, server side.
 * Helper functions for Select daemon.
 *
 * @author  Mikael Ylikoski
 * @date    2002
 */

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>
#include "protocol_s.h"
#include "utility.h"


/**
 * Server protocol data.
 */
struct protocol_s_data_ {
    char *ibuf;			/**< Input buffer */
    char *ibp;			/**< Input buffer pointer */
    int iblen;			/**< Input buffer length */
    int ilen;			/**< Currently used input buffer length */
    char *obuf;			/**< Output buffer */
    char *obp;			/**< Output buffer pointer */
    int oblen;			/**< Output buffer length */
    int sockfd;			/**< Socket file descriptor */
    struct sockaddr_un addr;	/**< Socket address */
};


/**
 * Create a new protocol session data structure.
 *
 * @param len  buffer size
 * @param adr  socket address
 * @return The new protocol session data.
 */
protocol_s_data *
protocol_s_new (int len, const char *adr) {
    int i;
    protocol_s_data *pd;

    pd = my_malloc (sizeof(protocol_s_data));
    pd->ibuf = my_malloc (len + 1);
    pd->ibp = pd->ibuf;
    pd->iblen = len;
    pd->ilen = 0;
    pd->obuf = my_malloc (len);
    pd->obp = pd->obuf;
    pd->oblen = len;
    pd->sockfd = -1;

    pd->addr.sun_family = AF_UNIX;
    memset (pd->addr.sun_path, 0, sizeof(pd->addr.sun_path));

    if (!adr)
	adr = "@my_select_classifier";

    i = strlen (adr);
    if (i > sizeof(pd->addr.sun_path) - 1)
	i = sizeof(pd->addr.sun_path) - 1;
    memcpy (pd->addr.sun_path, adr, i);
    if (pd->addr.sun_path[0] == '@')
	pd->addr.sun_path[0] = 0;

    return pd;
}


/**
 * Send a message.
 *
 * @param pd  protocol session data
 * @return Zero if ok, or nonzero otherwise.
 */
static int
protocol_send (protocol_s_data *pd) {
    int i, j;

    //*pd->obp = '\0';
    //printf ("out: %s\n", pd->obuf);

    j = pd->obp - pd->obuf;
    i = send (pd->sockfd, pd->obuf, j, 0);
    pd->obp = pd->obuf;
    if (i == j)
	return 0;

    fprintf (stderr, "Error: Cannot send message\n");
    return -1;
}


/**
 * Close the session.
 *
 * @param pd  protocol session data
 */
static int
protocol_close (protocol_s_data *pd) {
    close (pd->sockfd);
    pd->sockfd = -1;
    return 0;
}


/**
 * Reset input buffer for new session.
 *
 * @param pd  protocol session data
 * @param fd  file descriptor for new session
 */
void
protocol_s_reinit (protocol_s_data *pd, int fd) {
    pd->ibp = pd->ibuf;
    pd->ilen = 0;
    pd->sockfd = fd;
}


/**
 * Fill input buffer from socket.
 * Non-blocking.
 * May destroy anything before pd->ibp and anything after pd->ilen,
 * and change location of pd->ibp.
 *
 * @param pd  protocol session data
 * @return Zero if ok, or nonzero otherwise.
 */
static int
fill_buffer (protocol_s_data *pd) {
    int i;
    fd_set rfds;
    struct timeval tv;

    FD_ZERO(&rfds);
    FD_SET(pd->sockfd, &rfds);
    tv.tv_sec = 0;
    tv.tv_usec = 0;
    i = select (pd->sockfd + 1, &rfds, NULL, NULL, &tv);
    if (i < 1) {
	return -1;
    }

    if (pd->ibp > pd->ibuf) {
	memmove (pd->ibuf, pd->ibp, pd->ilen - (pd->ibp - pd->ibuf));
	pd->ilen -= pd->ibp - pd->ibuf;
	pd->ibp = pd->ibuf;
    }

    if (pd->ilen == pd->iblen) {
	pd->ibp = pd->ibuf;
	pd->ilen = 0;
    }

    i = recv (pd->sockfd, pd->ibuf + pd->ilen, pd->iblen - pd->ilen, 0);
    if (i < 1) {
	return -1;
    }
    pd->ilen += i;

    pd->ibuf[pd->ilen] = '\0';

    return 0;
}


/**
 * Receive a message on a server.
 *
 * @param pd  protocol session data
 * @return Zero if ok, or nonzero otherwise.
 */
int
protocol_s_receive (protocol_s_data *pd) {
    int i;

    if (pd->ibp - pd->ibuf < pd->ilen)
	return 0;

    pd->ibp = pd->ibuf;
    pd->ilen = 0;

    i = recv (pd->sockfd, pd->ibuf, pd->iblen, 0);
    if (i < 1) {
	fprintf (stderr, "Error: Cannot receive message\n");
	return -1;
    }
    pd->ilen += i;

    pd->ibuf[pd->ilen] = '\0';

    return 0;
}


/**
 * Read a part on a server.
 *
 * @param pd  protocol session data
 * @return The new part, or NULL if there was none available.
 */
static part *
protocol_read_part (protocol_s_data *pd) {
    char *ch;
    int i, j;
    part *p;

    p = my_malloc (sizeof(part));

    /* type */
    ch = strchr (pd->ibp, ':');
    if (!ch) {
	i = fill_buffer (pd);
	if (i) {
	    free (p);
	    return NULL;
	}
	ch = strchr (pd->ibp, ':');
	if (!ch) {
	    free (p);
	    return NULL;
	}
    }
    if (ch == pd->ibp) {	// No type specified, illegal
	free (p);
	return NULL;
    }
    *ch = '\0';
    p->type = my_strdup (pd->ibp);
    pd->ibp = ch + 1;

    /* charset */
    if (*pd->ibp == ':') {	// No charset specified, ok
	p->charset = NULL;
	pd->ibp++;
    } else {
	ch = strchr (pd->ibp, ':');
	if (!ch) {
	    i = fill_buffer (pd);
	    if (i) {
		free (p);
		return NULL;
	    }
	    ch = strchr (pd->ibp, ':');
	    if (!ch) {
		free (p);
		return NULL;
	    }
	}
	if (*pd->ibp == ':')	// No charset specified, ok
	    p->charset = NULL;
	else {
	    *ch = '\0';
	    p->charset = my_strdup (pd->ibp);
	}
	pd->ibp = ch + 1;
    }

    /* len */
    p->len = strtol (pd->ibp, &ch, 10);
    if (ch == pd->ibuf + pd->ilen) {	/* numbers all way to end of buffer */
	i = fill_buffer (pd);
	if (i) {
	    free (p->charset);
	    free (p);
	    return NULL;
	}
	p->len = strtol (pd->ibp, &ch, 10);
	if (ch == pd->ibuf + pd->ilen) {
	    free (p->charset);
	    free (p);
	    return NULL;
	}
    }
    if (ch == pd->ibp) {
	free (p->charset);
	free (p);
	return NULL;
    }

    /* ch now points to character after length and is within the buffer */
    if (*ch != ':') {
	free (p->charset);
	free (p);
	return NULL;
    }

    /* string */
    pd->ibp = ch + 1;		// ':'
    p->string = my_malloc (p->len + 1);
    for (j = 0; j < p->len; j += i) {
	if (pd->ibp - pd->ibuf == pd->ilen) {	// pd->ibp outside buffer
	    i = fill_buffer (pd);
	    if (i) {
		free (p->string);
		free (p->charset);
		free (p);
		return NULL;
	    }
	}
	i = p->len - j;
	if (i > pd->ilen - (pd->ibp - pd->ibuf)) {
	    i = pd->ilen - (pd->ibp - pd->ibuf);
	}
	memcpy (p->string + j, pd->ibp, i);
	pd->ibp += i;
    }
    p->string[p->len] = '\0';

    /* end marker */
    if (pd->ibp == pd->ibuf + pd->ilen) {	// pd->ibp outside buffer
	i = fill_buffer (pd);
	if (i) {
	    free (p->string);
	    free (p->charset);
	    free (p);
	    return NULL;
	}
    }
    if (*pd->ibp != '\n') {
	free (p->string);
	free (p->charset);
	free (p);
	return NULL;
    }
    pd->ibp++;	// '\n'

    return p;
}


/**
 * Read a request on a server.
 *
 * @param pd  protocol session data
 * @return The request, or NULL if there was none available.
 */
protocol_s_request *
protocol_s_read_request (protocol_s_data *pd) {
    char *str;
    int i;
    protocol_s_request *pr;

    pr = my_malloc (sizeof(protocol_s_request));
    pr->parts = NULL;
    pr->str = NULL;

    while (1) {
	if (!strncmp (pd->ibp, "open\n", 5)) {
	    pr->type = REQUEST_OPEN;
	    pd->ibp += 5;
	} else if (!strncmp (pd->ibp, "close\n", 6)) {
	    pr->type = REQUEST_CLOSE;
	    pd->ibp += 6;
	    protocol_close (pd);
	} else if (!strncmp (pd->ibp, "part:", 5)) {
	    pr->type = REQUEST_PARTS;
	    pd->ibp += 5;
	    pr->parts = protocol_read_part (pd);
	} else if (!strncmp (pd->ibp, "classify:", 9)) {
	    pr->type = REQUEST_CLASSIFY;
	    pd->ibp += 9;

	    /* find end */
	    str = strchr (pd->ibp, '\n');
	    if (!str) {
		i = fill_buffer (pd);
		if (i) {
		    free (pr);
		    return NULL;
		}
		str = strchr (pd->ibp, '\n');
		if (!str){
		    free (pr);
		    return NULL;
		}
	    }

	    pr->str = pd->ibp;

	    /* classify:*: */
	    str = strchr (pd->ibp, ':');
	    if (!str) {
		free (pr);
		return NULL;
	    }
	    pd->ibp = str;

	    /* classify:*:*\n */
	    str = strchr (pd->ibp, '\n');
	    *str = '\0';
	    *pd->ibp = '\0';
	    pd->ibp++;
	    pr->str2 = pd->ibp;
	    pd->ibp = str + 1;
	} else if (!strncmp (pd->ibp, "learn:", 6)) {
	    pr->type = REQUEST_LEARN;
	    pd->ibp += 6;

	    /* find end */
	    str = strchr (pd->ibp, '\n');
	    if (!str) {
		i = fill_buffer (pd);
		if (i) {
		    free (pr);
		    return NULL;
		}
		str = strchr (pd->ibp, '\n');
		if (!str){
		    free (pr);
		    return NULL;
		}
	    }
	    pr->str = pd->ibp;
	    *str = '\0';
	    pd->ibp = str + 1;
	} else if (!strncmp (pd->ibp, "get:", 4)) {
	    pr->type = REQUEST_GET;
	    pd->ibp += 4;

	    /* find end */
	    str = strchr (pd->ibp, '\n');
	    if (!str) {
		i = fill_buffer (pd);
		if (i) {
		    free (pr);
		    return NULL;
		}
		str = strchr (pd->ibp, '\n');
		if (!str){
		    free (pr);
		    return NULL;
		}
	    }
	    pr->str = pd->ibp;
	    *str = '\0';
	    pd->ibp = str + 1;
	} else if (!strncmp (pd->ibp, "set:", 4)) {
	    pr->type = REQUEST_SET;
	    pd->ibp += 4;

	    /* find end */
	    str = strchr (pd->ibp, '\n');
	    if (!str) {
		i = fill_buffer (pd);
		if (i) {
		    free (pr);
		    return NULL;
		}
		str = strchr (pd->ibp, '\n');
		if (!str){
		    free (pr);
		    return NULL;
		}
	    }

	    pr->str = pd->ibp;
	    *str = '\0';
	    pd->ibp = strchr (pd->ibp, ':');
	    *pd->ibp = '\0';
	    pd->ibp++;
	    pr->str2 = pd->ibp;
	    pd->ibp = str + 1;	// '\n'
	} else if (!strncmp (pd->ibp, "x-", 2)) {
	    pr->type = REQUEST_X;
	    // FIXME
	    str = strchr (pd->ibp, ':');
	    if (!str) {
		free (pr);
		return NULL;
	    }
	    pd->ibp = str + 1;

	    i = strtol (pd->ibp, &str, 10);
	    if (str == pd->ibp || pd->ibp - pd->ibuf + i + 2 >= pd->iblen ||
		*str != ':') {
		free (pr);
		return NULL;
	    }
	    pd->ibp = str + 1;

	    pr->str = pd->ibp;
	    pd->ibp += i;
	    if (*pd->ibp != '\n') {
		free (pr);
		return NULL;
	    }
	    *pd->ibp = '\0';
	    pd->ibp++;
	} else {
	    i = fill_buffer (pd);
	    if (i) {
		fprintf (stderr, "Error: Unknown request:%s:\n", pd->ibp);
		free (pr);
		return NULL;
	    }
	    continue;
	}
	break;
    }

    return pr;
}


/**
 * Make a rank list on a server.
 *
 * @param pd   protocol session data
 * @param il   array of ranks
 * @param len  length of il
 */
static int
protocol_rank (protocol_s_data *pd, int *il, int len) {
    int i;

    // FIXME check length
    pd->obp += sprintf (pd->obuf, "r:%d:", len);
    for (i = 0; i < len; i++)
	pd->obp += sprintf (pd->obp, "%d;", il[i]);
    pd->obp += sprintf (pd->obp, "\n");
    return 0;
}


/**
 * Make a score list on a server.
 *
 * @param pd   protocol session data
 * @param dl   array of scores
 * @param len  length of dl
 */
static int
protocol_score (protocol_s_data *pd, double *dl, int len) {
    int i;

    // FIXME check length
    pd->obp += sprintf (pd->obuf, "s:%d:", len);
    for (i = 0; i < len; i++)
	pd->obp += sprintf (pd->obp, "%.5f;", dl[i]);
    pd->obp += sprintf (pd->obp, "\n");
    return 0;
}


/**
 * Return a single rank classification from a server.
 *
 * @param pd  protocol session data
 * @param i   rank
 */
int
protocol_s_classify_top (protocol_s_data *pd, int i) {
    // FIXME check length
    pd->obp += sprintf (pd->obuf, "r:1:%d;\n", i);
    return protocol_send (pd);
}


/**
 * Return a rank classification from a server.
 *
 * @param pd   protocol session data
 * @param il   array of ranks
 * @param len  length of il
 */
int
protocol_s_classify_rank (protocol_s_data *pd, int *il, int len) {
    if (protocol_rank (pd, il, len))
	return -1;
    return protocol_send (pd);
}


/**
 * Return a score classification from a server.
 *
 * @param pd   protocol session data
 * @param dl   array of scores
 * @param len  length of dl
 */
int
protocol_s_classify_score (protocol_s_data *pd, double *dl, int len) {
    if (protocol_score (pd, dl, len))
	return -1;
    return protocol_send (pd);
}


/**
 * Write a string to the outbut buffer.
 *
 * @param pd   protocol session data
 * @param str  string to write
 * @return Zero if ok, or nonzero otherwise.
 */
int
protocol_s_write_string (protocol_s_data *pd, char *str) {
    int i;

    i = strlen (str);
    if (pd->obp - pd->obuf + i > pd->oblen)
	return -1;
    memcpy (pd->obp, str, i);
    pd->obp += i;
    return 0;
}


/**
 * Write an integer to the output buffer.
 *
 * @param pd  protocol session data
 * @param i   integer to write
 * @return Zero if ok, or nonzero otherwise.
 */
int
protocol_s_write_integer (protocol_s_data *pd, int i) {
    // FIXME check length
    pd->obp += sprintf (pd->obp, "%d", i);
    return 0;
}


/**
 * Send everything in the output buffer.
 *
 * @param pd  protocol data
 * @return Zero if ok, or nonzero otherwise.
 */
int
protocol_s_send (protocol_s_data *pd) {
    pd->obp += sprintf (pd->obp, "\n");
    return protocol_send (pd);
}


/**
 * Create server socket and accept connections indefinitely
 *
 * @param pd              protocol data
 * @param handle_session  session handler
 * @return Zero if ok, or nonzero otherwise.
 */
int
protocol_s_loop (protocol_s_data *pd, void (*handle_session) (int)) {
    int sockfd, newsock;

    sockfd = socket (PF_UNIX, SOCK_STREAM, 0);
    if (bind (sockfd, (struct sockaddr *)&pd->addr, sizeof(pd->addr))) {
	fprintf (stderr, "Error: Cannot bind address to socket\n");
	return -1;
    }
    if (listen (sockfd, 5)) {
	fprintf (stderr, "Error: Cannot listen for connections on socket\n");
	return -1;
    }

    while (1) {
	newsock = accept (sockfd, NULL, 0);
	if (newsock >= 0) {
	    handle_session (newsock);
	    close (newsock);
	}
    }

    return 0;
}