trace_ip_drv.c

来自「OTP是开放电信平台的简称」· C语言 代码 · 共 911 行 · 第 1/2 页

/* ``The contents of this file are subject to the Erlang Public License,
 * Version 1.1, (the "License"); you may not use this file except in
 * compliance with the License. You should have received a copy of the
 * Erlang Public License along with this software. If not, it can be
 * retrieved via the world wide web at http://www.erlang.org/.
 * 
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 * 
 * The Initial Developer of the Original Code is Ericsson Utvecklings AB.
 * Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
 * AB. All Rights Reserved.''
 * 
 *     $Id$
 */
/*
** This is a trace port, which means it cannot be busy and takes
** no opcodes.
** Send trace messages over the net.
*/
#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

#ifdef __WIN32__
#    include <winsock2.h>
#    include <windows.h>
#endif

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#ifndef __WIN32__
#    ifdef VXWORKS
#        include <sockLib.h>
#        include <sys/times.h>
#        include <iosLib.h>
#        include <taskLib.h>
#        include <selectLib.h>
#        include <ioLib.h>
#        include "reclaim.h"
#    endif 
#        include <unistd.h>
#        include <errno.h>
#        include <sys/types.h>
#        include <sys/socket.h>
#        include <netinet/in.h>
#        include <fcntl.h>
#endif

#ifdef DEBUG
#    ifndef __WIN32__
         /* erl_exit is not available to dll_drivers on windows. */
         void erl_exit(int, char *, ...);
#        define ASSERT(X)                               \
                    do {				\
                        if (!(X)) {			\
                            erl_exit(1,"%s",#X);	\
                        } 				\
                    } while(0) 
#    else 
#        include <assert.h>
#        define ASSERT(X) assert(X)
#    endif
#else 
#    define ASSERT(X)
#endif 

#define sock2event(s) ((ErlDrvEvent)(long)(s))
#define event2sock(p) ((SOCKET)(long)(p))

#include "erl_driver.h"

/*
** Protocol from driver:
** '\0' -> ok
** '\1' ++ String -> {error, Atom}
** Protocol when opening (arguments to start):
** <Portno> <Quesize> <Fl>
** Where...
** Portno, ascii string representing a number:
**     The TCP port number to listen to.
** Quesize, ascii string representing a number:
**     The number of messages to que up before dropping.
** Fl, ascii string representing a flag byte:
**     0x1 -> Drop oldest when que is full (instead of last arrived)
**     0x2 -> Fill the que even if noone is listening.
**
** The package sent over the network looks like this:
** +--+--------+-----------------------------------+
** |Op|Size NBO|Term in external format or empty   |
** +--+--------+-----------------------------------+
** Op, a char:
**    0 = binary, 1 = drop
**    If Op is 1, then Size reflects the number of dropped messages.
** Size, a 32 bit interger in network byte order:
**    Either the size of the binary term, or the number of packet's dropped. 
** Term, an array of bytes:
**    An erlang term in the external format or simply empty if Op == 1, the
**    term is Size long.
*/ 

/*
** SO, most of the differences between WinDoze and Posixish OS'es 
** is handeled here, but the multiplexing (driver_select) is also quite
** interesting, see my_driver_select further down in the file...
*/

#ifndef __WIN32__
#    define FLAG_READ DO_READ
#    define FLAG_WRITE DO_WRITE
     typedef int SOCKET;
#    define INVALID_SOCKET -1
#    define IS_INVALID_SOCKET(S) ((S) < 0)
#    define closesocket(Sock) close(Sock)
#    define ERRNO errno
#    ifdef EWOULDBLOCK
#        define ERRNO_BLOCK EWOULDBLOCK
#    else
#        ifdef EAGAIN
#            define ERRNO_BLOCK EAGAIN
#        else
#            error "No EWOULDBLOCK found"
#        endif
#    endif
#else /* Win32 */
#    define FLAG_READ (FD_READ | FD_CONNECT | FD_ACCEPT)
#    define FLAG_WRITE FD_WRITE
#    define ERRNO WSAGetLastError()
#    define ERRNO_BLOCK WSAEWOULDBLOCK
#    define IS_INVALID_SOCKET(S) ((S) == INVALID_SOCKET)
#endif


/*
** Option flags
*/
#define FLAG_DROP_OLDEST    1
#define FLAG_FILL_ALWAYS    2
#define FLAG_LISTEN_PORT    4

/*
** Op's in messages
*/
#define OP_BINARY 0
#define OP_DROP   1
     
/*
** State structure
*/

typedef struct trace_ip_message {
    int siz; /* the size of the "binary data" */
    int written; /* if only a part was written, when == siz, all is written */
    unsigned char bin[1]; /* The opcode, the Size and optionally the binary. */
} TraceIpMessage;
    

typedef struct trace_ip_data {
    unsigned flags;
    int listen_portno;
    SOCKET listenfd;
    SOCKET fd;
#ifdef __WIN32__
    unsigned listen_event_mask;
    HANDLE listen_event;
    unsigned event_mask;
    HANDLE event;
#endif
    ErlDrvPort port;
    struct trace_ip_data *next;
    int quesiz;
    int questart;
    int questop;
    TraceIpMessage *que[1]; /* You guessed it, will be longer... */
} TraceIpData;

static TraceIpData *first_data; 

/*
** Interface routines
*/
static ErlDrvData trace_ip_start(ErlDrvPort port, char *buff);
static void trace_ip_stop(ErlDrvData handle);
static void trace_ip_output(ErlDrvData handle, char *buff, int bufflen);
#ifdef __WIN32__
static void trace_ip_event(ErlDrvData handle, ErlDrvEvent event);
#endif
static void trace_ip_ready_input(ErlDrvData handle, ErlDrvEvent fd);
static void trace_ip_ready_output(ErlDrvData handle, ErlDrvEvent fd);
static void trace_ip_finish(void); /* No arguments, despite what might be stated
				     in any documentation */
static int trace_ip_control(ErlDrvData handle, unsigned int command, 
			    char* buff, int count, 
			    char** res, int res_size);

/*
** Internal routines
*/
static void *my_alloc(size_t size);
static ErlDrvBinary *my_alloc_binary(int size);
static int write_until_done(SOCKET s, char *buff, int bufflen);
static unsigned get_be(unsigned char *s);
static void put_be32(unsigned n, unsigned char *s);
static void put_be16(unsigned n, unsigned char *s);
static TraceIpData *lookup_data_by_port(int portno);
static int set_nonblocking(SOCKET sock);
static TraceIpMessage *make_buffer(int datasiz, unsigned char op, 
				   unsigned number);
static void enque_message(TraceIpData *data, unsigned char *buff, int bufflen,
			  int byteswritten);
static void clean_que(TraceIpData *data);
static void close_client(TraceIpData *data);
static int trywrite(TraceIpData *data, unsigned char *buff, int bufflen);
static SOCKET my_accept(SOCKET sock);
static void close_unlink_port(TraceIpData *data);
static int my_driver_select(TraceIpData *desc, SOCKET fd, int flags, int on);
/*
** The driver struct
*/
ErlDrvEntry trace_ip_driver_entry = {
    NULL,	           /* F_PTR init, N/A */
    trace_ip_start,        /* L_PTR start, called when port is opened */
    trace_ip_stop,         /* F_PTR stop, called when port is closed */
    trace_ip_output,       /* F_PTR output, called when erlang has sent */
#ifdef __WIN32__
    trace_ip_event,        /* Anything happens on an associated event */
    NULL,                  /* Write selections not supported on win32 */
#else
    trace_ip_ready_input,  /* F_PTR ready_input, called when input descriptor 
			      ready */
    trace_ip_ready_output, /* F_PTR ready_output, called when output 
			      descriptor ready */
#endif
    "trace_ip_drv",        /* char *driver_name, the argument to open_port */
    trace_ip_finish,       /* F_PTR finish, called when unloaded */
    NULL,                  /* void * that is not used */
    trace_ip_control,      /* F_PTR control, port_control callback */
    NULL,                  /* F_PTR timeout, reserved */
    NULL                   /* F_PTR outputv, reserved */
};

/*
** Driver initialization routine
**
** No matter what's stated otherwise, this function shall return a pointer.
*/

DRIVER_INIT(trace_ip_drv)
{
    first_data = NULL;
    /*trace_ip_driver_entry.handle = handle; ??? What is this, and why? It is no more! */
    return &trace_ip_driver_entry;
}

/*
** Driver interface routines
*/

/*
** Open a port
*/
static ErlDrvData trace_ip_start(ErlDrvPort port, char *buff)
{
    TraceIpData *ret;
    int portno;
    int quesiz;
    int flags;
    SOCKET s;
    struct sockaddr_in sin;
    int reuse = 1;

#ifdef HARDDEBUG
    fprintf(stderr,"trace_ip_drv/trace_ip_start (%s)\r\n", buff);
#endif
    if (sscanf(buff,"trace_ip_drv %d %d %d",&portno, &quesiz, &flags) != 3)
	return ERL_DRV_ERROR_GENERAL;

    if (flags > 3 || flags < 0 || portno < 0 || quesiz < 0)
	return ERL_DRV_ERROR_GENERAL;
    
    if (lookup_data_by_port(portno) != NULL)
	return ERL_DRV_ERROR_GENERAL;
    
    if (IS_INVALID_SOCKET(s = socket(AF_INET, SOCK_STREAM, 0)))
	return ERL_DRV_ERROR_GENERAL;

    if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, 
		   (char *) &reuse, sizeof(reuse)) < 0) {
	closesocket(s);
	return ERL_DRV_ERROR_GENERAL;
    }


    memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = INADDR_ANY;
    sin.sin_port = htons((short) portno);
    
    if (bind(s, (struct sockaddr *)&sin, sizeof(sin)) != 0) {
	closesocket(s);
	return ERL_DRV_ERROR_GENERAL;
    }

    if (portno == 0) {
#ifdef HAVE_SOCKLEN_T
	socklen_t sinlen = sizeof(sin);
#else
	int  sinlen = (int) sizeof(sin);
#endif
	if (getsockname(s, (struct sockaddr *)&sin, &sinlen) != 0) {
	    closesocket(s);
	    return ERL_DRV_ERROR_GENERAL;
	} else {
	    portno = ntohs(sin.sin_port);
	}
    }    

    if (listen(s, 1)) { /* No significant backlog needed */
	closesocket(s);
	return ERL_DRV_ERROR_GENERAL;
    }

    if (set_nonblocking(s) != 0){
	closesocket(s);
	return ERL_DRV_ERROR_GENERAL;
    }
    
    /* OK, the socket is created, lets build the structure. */
    /* Deliberately one more pointer than the quesize specified... */
    ret = my_alloc(sizeof(TraceIpData) + 
		   quesiz * sizeof(TraceIpMessage *));
    
    ret->flags = flags | FLAG_LISTEN_PORT;
    ret->listen_portno = portno;
    ret->listenfd = s;
    ret->fd = INVALID_SOCKET;
    ret->port = port;
    ret->next = first_data;
    ret->quesiz = quesiz+1;
    ret->questart = ret->questop = 0;
    memset(ret->que, 0, ret->quesiz);
    
    first_data = ret;
#ifdef __WIN32__
    ret->listen_event_mask = 0;
    ret->listen_event = 0;
    ret->event_mask = 0;
    ret->event = 0;
#endif
    my_driver_select(ret, s, FLAG_READ, 1);
    set_port_control_flags(port, PORT_CONTROL_FLAG_BINARY);

    return (ErlDrvData) ret;
}

/*
** Close a port
*/
static void trace_ip_stop(ErlDrvData handle)
{
    close_unlink_port((TraceIpData *) handle);
}

/*
** Data sent from erlang to port.
*/
static void trace_ip_output(ErlDrvData handle, char *buff, int bufflen)
{
    TraceIpData *data = (TraceIpData *) handle;
    if (data->flags & FLAG_LISTEN_PORT) {
	if (data->flags & FLAG_FILL_ALWAYS) {
	    enque_message(data, buff, bufflen, 0);
	}
	return;
    }
    if (data->que[data->questart] != NULL) {
	trace_ip_ready_output(handle, sock2event(data->fd));
    }
    if (data->que[data->questart] == NULL) {
	int written = trywrite(data, buff, bufflen);
	if (written >= 0 && written < bufflen + 5) {
	    enque_message(data, buff, bufflen, written);
	    my_driver_select(data, data->fd, FLAG_WRITE, 1);
	}
	return;
    }
    enque_message(data, buff, bufflen, 0);
    return;
}

/*
** We have something to read from a file descriptor
*/
static void trace_ip_ready_input(ErlDrvData handle, ErlDrvEvent fd)
{
    TraceIpData *data = (TraceIpData *) handle;
    int client;

    if (!(data->flags & FLAG_LISTEN_PORT) && event2sock(fd) == data->listenfd) {
	/*
	** Someone tries to connect to already connected port, 
	** just accept and close.
	*/
	if (!IS_INVALID_SOCKET((client = my_accept(data->listenfd)))) {
	    closesocket(client);
	}
	return;
    }

    if (event2sock(fd) == data->listenfd) {
	/*
	** Maybe accept, we are a listen port...
	*/
	if (!IS_INVALID_SOCKET((client = my_accept(data->listenfd)))) {
	    data->fd = client;
	    set_nonblocking(client);
	    if (data->que[data->questart] != NULL) {
		my_driver_select(data, data->fd, FLAG_WRITE | FLAG_READ, 1);
	    } else {
		my_driver_select(data, data->fd, FLAG_READ, 1);
	    }
	    data->flags &= ~(FLAG_LISTEN_PORT);
	}
	return;
    }

    /*
     * It is a probably EOF because the other end closed the socket,
     * but better make sure.
     */

    if ((SOCKET)fd == data->fd) {
#ifdef __WIN32__
	close_client(data);
#else
	int res;
	char sbuf[128];
	
	if ((res = read(data->fd, sbuf, sizeof sbuf)) == 0) {
	    close_client(data);
	}

	/*
	 * Something else. Just ignore it.
	 *
	 * When /dev/poll is used on Solaris, this callback can
	 * be called even if there is nothing to read. An attempt
	 * to read will result in an EAGAIN error.
	 */
#ifdef DEBUG
	if (res < 0) {
	    fprintf(stderr, "Read on fd %d failed with errno=%d\r\n",
		    data->fd, errno);