inet_gethost.c

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

/*
 * Erlang port program to do the name service lookup for the erlang 
 * distribution and inet part of the kernel.
 * A pool of subprocess is kept, to which a pair of pipes is connected.
 * The main process schedules requests among the different subprocesses 
 * (created with fork()), to be able to handle as many requests as possible
 * simultaneously. The controlling erlang machine may request a "cancel",
 * in which case the process may be killed and restarted when the need arises.
 * The single numeric parameter to this program is the maximum port pool size,
 * which is the size of the bookkeeping array.
 */ 

#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

#ifdef WIN32

#define WIN32_LEAN_AND_MEAN
#include <winsock2.h>
#include <windows.h>
#include <process.h>
#include <stdio.h>
#include <stdlib.h>

#else /* Unix */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <stdarg.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <errno.h>
#include <signal.h>

#ifdef HAVE_SYS_TIME_H
#include <sys/time.h>
#else
#include <time.h>
#endif
#include <sys/times.h>

#ifndef RETSIGTYPE
#define RETSIGTYPE void
#endif

#endif /* !WIN32 */

#define PACKET_BYTES 4
#define READ_PACKET_BYTES(X,Y) read_int32((X),(Y))
#define PUT_PACKET_BYTES(X,Y) put_int32((X),(Y)) 
/* The serial numbers of the requests */
typedef int SerialType;

#define INVALID_SERIAL -1

/* The operations performed by this program */
typedef unsigned char OpType;

#define OP_GETHOSTBYNAME 1
#define OP_GETHOSTBYADDR 2
#define OP_CANCEL_REQUEST 3
#define OP_CONTROL 4

/* The protocol (IPV4/IPV6) */
typedef unsigned char ProtoType;

#define PROTO_IPV4 1
#define PROTO_IPV6 2

/* OP_CONTROL */
typedef unsigned char CtlType;
#define SETOPT_DEBUG_LEVEL 0

/* The unit of an IP address (0 == error, 4 == IPV4, 16 == IPV6) */
typedef unsigned char UnitType;

#define UNIT_ERROR 0
#define UNIT_IPV4  4
#define UNIT_IPV6 16

/* And the byte type */
typedef unsigned char AddrByte; /* Must be compatible with character 
				   datatype */

/* 
 * Marshalled format of request: 
 *{
 *  Serial: 32 bit big endian
 *  Op:8 bit  [1,2,3]
 *  If op == 1 {
 *    Proto:8 bit [1,2]
 *    Str: Null terminated array of characters
 *  } Else if op == 2 {
 *    Proto:8 bit [1,2]
 *    If proto == 1 {
 *      B0..B3: 4 bytes, most significant first
 *    } Else (proto == 2) {
 *      B0..B15: 16 bytes, most significant first
 *    }
 *  } 
 *  (No more if op == 3)
 *} 
 * The request arrives as a packet, with 4 packet size bytes.
 */

/* The main process unpackes the marshalled message and sends the data 
 * to a suitable port process or, in the case of a close request, kills the
 * suitable port process. There is also a que of requests linked together,
 * for when all subrocesses are busy.
 */

typedef struct QueItem {
    struct QueItem *next;
    int req_size;
    AddrByte request[1];
} QueItem; /* Variable size due to request's variable size */

QueItem *que_first;
QueItem *que_last;

#ifdef WIN32
typedef struct mesq {
    HANDLE data_present;
    CRITICAL_SECTION crit;
    int shutdown;
    QueItem *first;
    QueItem *last;
} MesQ;

MesQ *to_erlang;
MesQ *from_erlang;
#endif

/*
 * Marshalled format of reply:
 *{
 *  Serial: 32 bit big endian
 *  Unit: 8 bit, same as h_length or 0 for error
 *  if unit == 0 {
 *    Str: Null terminated character string explaining the error
 *  } else {
 *    Naddr: 32 bit big endian
 *    if unit = 4 {
 *      (B0..B3)0..(B0..B3)Naddr-1: Naddr*4 bytes most significant first
 *    } else if unit == 16 {
 *      (B0..B15)0..(B0..B15)Naddr-1: Naddr*16 bytes most significant first
 *    }
 *    Nnames: 32 bit big endian >= 1
 *    Name0: Null terminated string of characters
 *    Alias[0]..Alias[Nnames - 2]: Nnames - 1 Null terminated strings of chars
 *  }
 *}
 * Four packet size bytes prepended (big endian)
 */
/* Internal error codes */
#define ERRCODE_NOTSUP 1
#define ERRCODE_HOST_NOT_FOUND 2
#define ERRCODE_TRY_AGAIN 3
#define ERRCODE_NO_RECOVERY 4
#define ERRCODE_NO_DATA 5
#define ERRCODE_NETDB_INTERNAL 7

/* 
 * Each worker process is represented in the parent by the following struct
 */

typedef unsigned WorkerState;

#define WORKER_EMPTY   0 /* No process created       */
#define WORKER_FREE    1 /* Living waiting process   */
#define WORKER_BUSY    2 /* Living busy process      */
#define WORKER_STALLED 3 /* Living cancelled process */

/* The timeout when killing a child process in seconds*/
#define CHILDWAIT_TMO 1
/* The domainname size_limit */
#define DOMAINNAME_MAX 258 /* 255 + Opcode + Protocol + Null termination */

typedef struct {
    WorkerState state;
#ifdef WIN32
    DWORD pid; /* 0 if unused */
    MesQ *writeto; /* Message queues */
    MesQ *readfrom;
#else
    pid_t pid; /* -1 if unused */
    int writeto, readfrom; /* Pipes */
#endif
    SerialType serial;
    AddrByte domain[DOMAINNAME_MAX];
    QueItem *que_first;
    QueItem *que_last;
    int que_size;
} Worker;

int num_busy_workers;
int num_free_workers;
int num_stalled_workers;
int max_workers;
int greedy_threshold;
Worker *busy_workers;  /* Workers doing any job that someone really is 
			  interested in */
Worker *free_workers;  /* Really free workers */
Worker *stalled_workers; /* May still deliver answers which we will 
			    discard */
#define BEE_GREEDY() (num_busy_workers >= greedy_threshold)

static char *program_name;

static int debug_level;

#ifdef NODEBUG
#define DEBUGF(L,P) /* Nothing */
#else
#define DEBUGF(Level,Printf) do { if (debug_level >= (Level)) \
                                      debugf Printf;} while(0)  
#endif
#define ALLOC(Size) my_malloc(Size)
#define REALLOC(Old, Size) my_realloc((Old), (Size))
#define FREE(Ptr) free(Ptr)

#ifdef WIN32
#define WAKEUP_WINSOCK() do {			\
    char dummy_buff[100];			\
    gethostname(dummy_buff,99);			\
} while (0)
#endif

/* The internal prototypes */
static char *format_address(int siz, AddrByte *addr);
static void debugf(char *format, ...);
static void warning(char *format, ...);
static void fatal(char *format, ...);
static void *my_malloc(size_t size);
static void *my_realloc(void *old, size_t size);
static int get_int32(AddrByte *buff);
static void put_int32(AddrByte *buff, int value);
static int create_worker(Worker *pworker, int save_que);
static int map_netdb_error(int netdb_code);
static char *errcode_to_string(int errcode);
static size_t build_error_reply(SerialType serial, int errnum, 
				AddrByte **preply,
				size_t *preply_size);
static size_t build_reply(SerialType serial, struct hostent *he, 
			  AddrByte **preply, size_t *preply_size);
static int read_request(AddrByte **buff, size_t *buff_size);
static OpType get_op(AddrByte *buff);
static AddrByte *get_op_addr(AddrByte *buff);
static SerialType get_serial(AddrByte *buff);
static ProtoType get_proto(AddrByte *buff);
static CtlType get_ctl(AddrByte *buff);
static AddrByte *get_data(AddrByte *buff);
static int get_debug_level(AddrByte *buff);
static int relay_reply(Worker *pw);
static int ignore_reply(Worker *pw);
static void init_workers(int max);
static void kill_worker(Worker *pw);
static Worker *pick_worker(void);
static void kill_last_picked_worker(void);
static void stall_worker(SerialType serial);
static int handle_io_busy(int ndx);
static int handle_io_free(int ndx);
static int handle_io_stalled(int ndx);
static void check_que(void);
static void main_loop(void);
static void usage(char *unknown);
static void domaincopy(AddrByte *out,AddrByte *in);
static int domaineq(AddrByte *d1, AddrByte *d2);
static int get_domainname(AddrByte *inbuff, int insize, AddrByte *domainbuff);
static Worker *pick_worker_greedy(AddrByte *domainbuff);
static void restart_worker(Worker *w);
static void start_que_request(Worker *w) ;
#ifdef WIN32
static int read_int32(HANDLE fd, int *res);
static int read_exact(HANDLE fd, void *vbuff, DWORD nbytes);
static int write_exact(HANDLE fd, AddrByte *buff, DWORD len);
DWORD WINAPI worker_loop(void *v);
DWORD WINAPI reader(void *data);
DWORD WINAPI writer(void *data);
static int send_mes_to_worker(QueItem *m, Worker *pw);
BOOL create_mesq(MesQ **q);
BOOL enque_mesq(MesQ *q, QueItem *m);
BOOL deque_mesq(MesQ *q, QueItem **m);
BOOL close_mesq(MesQ *q);
HANDLE event_mesq(MesQ *q);
#else
static size_t read_int32(int fd, int *res);
static ssize_t read_exact(int fd, void *vbuff, size_t nbytes);
static int write_exact(int fd, AddrByte *buff, int len);
void reap_children(int ignored);
static void init_signals(void);
static void kill_all_workers(void);
static void close_all_worker_fds(void);
static int worker_loop(void);
static int fillin_reply(Worker *pw);
static int send_request_to_worker(AddrByte *pr, int rsize, Worker *pw);
#endif

/*
 * Main
 */
int main(int argc, char **argv) 
{
    int num_workers = 1;
    char *dls;
    char **ap = argv + 1;
    int x;
    int disable_greedy = 0;

    program_name = *argv;
    que_first = que_last = NULL;
    if ((dls = getenv("ERL_INET_GETHOST_DEBUG")) != NULL) {
	debug_level = atoi(dls);
    } else {
	debug_level = 0;
    }
    greedy_threshold = 0;

    while (*ap) {
	if (!strcmp(*ap, "-d")) {
	    ++debug_level;
	} else if(!strcmp(*ap, "-g") && *(ap + 1)) {
	    ++ap;
	    x = atoi(*ap);
	    if (!x) {
		usage(*ap);
	    } else {
		greedy_threshold = x;
	    }
	} else if(!strcmp(*ap, "-ng")) {
	    disable_greedy = 1;
	} else {
	    x = atoi(*ap);
	    if (!x) {
		usage(*ap);
	    } else {
		num_workers = x;
	    }
	}
	++ap;
    }

#ifdef WIN32
    if (num_workers > 60 || greedy_threshold > 60) {
	usage("More than 60 workers on windows impossible!");
	num_workers = 60;
	greedy_threshold = 0;
    }
#endif

    if(!greedy_threshold) {
	greedy_threshold = (3*num_workers)/4; /* 75% */
	if (!greedy_threshold) {
	    greedy_threshold = num_workers;
	}
    }

    if (disable_greedy) {
	greedy_threshold = num_workers + 1;
    }

#ifdef WIN32
    {
	WORD wr;
	WSADATA wsa_data;
	int wsa_error;
	wr = MAKEWORD(2,0);
	
	wsa_error = WSAStartup(wr,&wsa_data);
	if (wsa_error) {
	    fatal("Could not open usable winsock library.");
	}
	if (LOBYTE(wsa_data.wVersion) != 2 || HIBYTE(wsa_data.wVersion) != 0) {
	    fatal("Could not open recent enough winsock library.");
	}

	if (debug_level >= 1) {
	    AllocConsole();
	    DEBUGF(1,("num_workers = %d, greedy_threshold = %d, "
		      "debug_level = %d.",
		      num_workers, greedy_threshold, debug_level));
	}
    }
    WAKEUP_WINSOCK(); /* Why on earth is this needed? */
#endif

    init_workers(num_workers);
    main_loop();
#ifndef WIN32
    kill_all_workers();
#endif
    return 0;
}

static void usage(char *unknown)
{
    fprintf(stderr,"%s: Unknown option \"%s\"\n"
	    "Usage: %s [-d [-d ...]] [-g <greedy threshold>] "
	    "[<number of workers>]\n",
	    program_name, unknown, program_name);
}

/*
 * Main process main loop
 */

static int handle_io_busy(int ndx)
{
    /* Probably an answer */
    int res;
    res = relay_reply(&busy_workers[ndx]);
    if (res < 0) {
	/* Bad worker */
	if (busy_workers[ndx].que_size) {
	    restart_worker(&busy_workers[ndx]);
	    start_que_request(&busy_workers[ndx]);
	    return 0;
	} else {
	    kill_worker(&busy_workers[ndx]);
	    --num_busy_workers;
	    busy_workers[ndx] = busy_workers[num_busy_workers];
	}
	return 1; 
    } else if (res == 0) {
	/* Erlang has closed */
	return -1;
    } else {
	if (busy_workers[ndx].que_size) {
	   start_que_request(&busy_workers[ndx]);
	   return 0;
	}
	/* The worker is no longer busy, it should be in the free list */
	free_workers[num_free_workers] = busy_workers[ndx];
	free_workers[num_free_workers].state = WORKER_FREE;
	++num_free_workers;
	--num_busy_workers;
	busy_workers[ndx] = busy_workers[num_busy_workers];
	return 1;
    }
}

static int handle_io_free(int ndx)
{
    /* IO from a free worker means "kill me" */
    DEBUGF(1,("Free worker[%ld] spontaneously died.",
	      (long) free_workers[ndx].pid));
    kill_worker(&free_workers[ndx]);
    --num_free_workers;
    free_workers[ndx] = free_workers[num_free_workers];
    return 1;
}

static int handle_io_stalled(int ndx)
{
    int res;
    res = ignore_reply(&stalled_workers[ndx]);
    if (res <= 0) {
	/* Bad worker */
	kill_worker(&stalled_workers[ndx]);
	--num_stalled_workers;
	stalled_workers[ndx] = stalled_workers[num_stalled_workers];
	return 1; 
    } else {
	DEBUGF(3,("Ignoring reply from stalled worker[%ld].",
		  (long) stalled_workers[ndx].pid)); 
	free_workers[num_free_workers] = stalled_workers[ndx];
	free_workers[num_free_workers].state = WORKER_FREE;
	++num_free_workers;
	--num_stalled_workers;
	stalled_workers[ndx] = stalled_workers[num_stalled_workers];
	return 1;
    }
}

static void check_que(void) 
{
    /* Check if anything in the que can be handled */
    Worker *cw;

    while (que_first) {
	QueItem *qi;
	if ((cw = pick_worker()) == NULL) {
	    break;
	}
#ifdef WIN32
	{
	    SerialType save_serial = get_serial(que_first->request);
	    if (send_mes_to_worker(que_first, cw) != 0) {
		kill_last_picked_worker();
		continue;
	    }
	    cw->serial = save_serial;
	}
#else	    
	if (send_request_to_worker(que_first->request, 
				   que_first->req_size, cw) != 0) {
	    /* Couldn't send request, kill the worker and retry */
	    kill_last_picked_worker();
	    continue;
	}
	cw->serial = get_serial(que_first->request);
#endif
	/* Went well, lets deque */
	qi = que_first;
	que_first = qi->next;
	if (que_first == NULL) {
	    que_last = NULL;
	}
	DEBUGF(3,("Did deque serial %d, Que is %sempty",
		  get_serial(qi->request), (que_first) ? "not " : ""));
#ifndef WIN32
	FREE(qi);
#endif
    }
}

static int clean_que_of(SerialType s)
{
    QueItem **qi;
    int i;

    for(qi=&que_first;*qi != NULL && 
	    s != get_serial((*qi)->request); qi = &((*qi)->next))
	;
    if(*qi != NULL) {
	QueItem *r = *qi;
	*qi = (*qi)->next;
	FREE(r);
	if(que_last == r) {
	    /* Lost the "last" pointer, should be very uncommon
	       if the que is not empty, so we simply do a traversal
	       to reclaim it. */
	    if (que_first == NULL) {
		que_last = NULL;
	    } else {
		for (que_last=que_first;que_last->next != NULL;
		     que_last = que_last->next)
		    ;
	    }
	}
	DEBUGF(3,("Removing serial %d from global que on request, "
		  "que %sempty",s, (que_first) ? "not " : ""));
	return 1;
    }
    for (i = 0; i < num_busy_workers; ++i) {
	for(qi=&(busy_workers[i].que_first);*qi != NULL && 
		s != get_serial((*qi)->request); qi = &((*qi)->next))
	    ;
	if(*qi != NULL) {
	    QueItem *r = *qi;
	    *qi = (*qi)->next;
	    FREE(r);
	    if(busy_workers[i].que_last == r) {
		/* Lost the "last" pointer, should be very uncommon
		   if the que is not empty, so we simply do a traversal
		   to reclaim it. */
		if (busy_workers[i].que_first == NULL) {
		    busy_workers[i].que_last = NULL;
		    if (busy_workers[i].que_size != 1) {
			fatal("Worker que size counter incorrect, internal datastructure error.");
		    }
		} else {
		    for (busy_workers[i].que_last = busy_workers[i].que_first;
			 busy_workers[i].que_last->next != NULL;
			 busy_workers[i].que_last = busy_workers[i].que_last->next)
			;
		}
	    }
	    --(busy_workers[i].que_size);
	    DEBUGF(3,("Removing serial %d from worker[%ld] specific que "
		      "on request, que %sempty", 
		      s, (long) busy_workers[i].pid, 
		      (busy_workers[i].que_first) ? "not " : ""));
	    return 1;