core.h

这是基于C语言开发的分布式搜索源代码

#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <stdlib.h>
#include <stdint.h>
#include <assert.h>
#ifdef HAVE_GETOPT_H
#include <getopt.h>
#endif
#include <locale.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <sys/mman.h>
#ifdef HAVE_SYS_FILIO_H
#include <sys/filio.h>
#endif
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#ifdef HAVE_PTHREAD_H
#include <pthread.h>
#endif
#include <evbase.h>
#include "sbase.h"
#include "log.h"
#include "buffer.h"
#include "chunk.h"
#include "timer.h"

#ifndef _CORE_H
#define _CORE_H
#ifdef __cplusplus
extern "C" {
#endif

/* INET */
#ifndef TCP_T
#define TCP_T SOCK_STREAM
#endif
#ifndef UDP_T	
#define UDP_T SOCK_DGRAM 
#endif
#ifndef SOCK_MIX      
#define SOCK_MIX (SOCK_DGRAM | SOCK_STREAM)
#endif
#ifndef DGRAM_SIZE 	
#define DGRAM_SIZE 1024 
#endif
#ifndef BUF_SIZE
#define	BUF_SIZE   1024 * 1024
#endif
#ifndef MAX_CONNECTIONS
#define MAX_CONNECTIONS 	65535
#endif
#ifndef MAX_PACKET_LEN
#define MAX_PACKET_LEN		65535
#endif
#ifndef MAX_THREADS
#define MAX_THREADS		256
#endif
#ifndef MIN_SLEEP_USEC
#define MIN_SLEEP_USEC		1000
#endif 
#ifndef MIN_HEARTBEAT_INTERVAL
#define MIN_HEARTBEAT_INTERVAL  10
#endif
#ifndef MIN_CONN_TIMEOUT
#define MIN_CONN_TIMEOUT	60
#endif
#ifndef  SESSION_MAP_SIZE
#define  SESSION_MAP_SIZE 	1024
#endif
#ifndef	 IP_MAX
#define  IP_MAX			16
#endif
#ifndef	INET_BACKLOG
#define	INET_BACKLOG			65535 
#endif


/* MESSAGE DEFINE*/
#ifndef TYPEDEF_MESSAGE
#define TYPEDEF_MESSAGE

/* MESSAGE ACTION ID DEFINE */
#define MESSAGE_QUIT            0x00
#define MESSAGE_NEW_SESSION     0x01
#define MESSAGE_INPUT           0x02
#define MESSAGE_OUTPUT          0x04
#define MESSAGE_PACKET          0x08
#define MESSAGE_DATA          	0x10

typedef struct _MESSAGE{
	int		msg_id;
	int		fd;	
	void		*handler;
	
	void		(*clean)(struct _MESSAGE **);
}MESSAGE;
MESSAGE *message_init();
/* Clean message */
void message_clean(MESSAGE **msg);
#define MESSAGE_SIZE	sizeof(MESSAGE)
#endif


/* TYPES PREDEFINE */
struct _SERVER;
struct _THREAD;
struct _SESSION;
struct _QUEUE;
struct _BUFFER;

/* STRUCTS DEFINE */
#ifndef TYPEDEF_SERVER
#define TYPEDEF_SERVER
typedef struct _SERVER{
	SBASE *sb;
	/* base setting */
	int			running_status;
	int			sock_t; /* socket type*/
	int			tcpfd;
	int			udpfd;	
	int   			domain;
        int   			family;
        char   			*ip;
        int   			port;
        int   			backlog;
	struct sockaddr_in 	lsa;

	/* global log pointer */
	LOG	*log;
	/* timer */
	TIMER 	*timer;

	/* SERVER TYPE */
	int			server_type ;
	/* customize  define */
	int			packet_t;
	int 			packet_len ;
	void			*delimiter;
	int			delimiter_len;
	int			buf_size;
	/* Usleep Setting*/
	uint32_t            	heartbeat_interval ;
        uint32_t            	sleep_usec ;
        uint32_t            	conn_timeout ;

	/* hook funtions for SESSION */
	/* Heartbeat Handler */
        void (*heartbeat_handler)(void);
        /* Read From Buffer and return packet length to get */
        int (*packet_reader)(const HANDLER *, const BUFFER *);
	/* Packet Handler */
        void (*packet_handler)(const HANDLER *, const BUFFER *);
	/* Data Handler */
        void (*data_handler)(const HANDLER *, const BUFFER *, const CHUNK *, const BUFFER *);
	/* OOB Data Handler */
        void (*oob_handler)(const HANDLER *handler, const BUFFER *oob);

	/* thread setting */
	int			max_threads; /* max thread limitation */
	int			running_threads;
	struct _THREAD 		**threads; /* point to threads pool */
	
	/* event setting */
	//struct event		sv_event;
	//struct event_base	*eventbase;
	EVENT	*sv_event;
	EVBASE  *eventbase;
	
	/* connection setting */
	int			max_connections; /* max connection limitation */
	int 			running_connections; /* connection number */

	/* methods */
	void (*start)(struct _SERVER * );		
	void (*stop)(struct _SERVER * );		
        void (*event_handler)(int, short, void*);
        int  (*init)(struct _SERVER * );
        void (*run)(struct _SERVER * );
        int  (*addconn)(struct _SERVER *, int , struct sockaddr_in);
        void (*terminate)(struct _SERVER * );
	void (*clean)(struct _SERVER **);
}SERVER;
#endif

#ifndef TYPEDEF_THREAD
#define TYPEDEF_THREAD
typedef struct _THREAD{
	/* server pointer and hook */
	struct _SERVER 		*sv;

	/* global log pointer */
        LOG     *log;
	
	/* base setting */
	pthread_mutex_t 	mutex;
	int			index;  /* index ID in threads pool */		
	pthread_t 		thread_id; /* thread_id */
  	int			running_status; /* show running status*/

	/* libevent */
	//struct event_base  	*eventbase;
	EVBASE 	 *eventbase;
	
	/* message queue setting */
	struct _QUEUE 		*message_queue;
	
	/* session setting */
	struct _SESSION  	**sessions;
	
	/* Timer */	
	TIMER *timer ;

	/* methods */
	/* THREAD event handler */
	void	 (*event_handler)(int, short , void *);
       	void* 	 (*run)(void *);
        int	 (*addconn)(struct _THREAD *, int , struct sockaddr_in);
        int	 (*add_session)(struct _THREAD *, int, struct sockaddr_in);
        void	 (*terminate_session)(struct _THREAD *, struct _SESSION *);
        void	 (*state_conns)(struct _THREAD *);
        void	 (*terminate)(struct _THREAD *);
	void	 (*clean)(struct _THREAD **);
}THREAD;
#endif

#ifndef TYPEDEF_SESSION
#define TYPEDEF_SESSION
typedef struct _SESSION{
	/* thread pointer */
        struct _THREAD          *pth;

	/* base setting */	
	int			fd;
	struct sockaddr_in	sa;
	char			ip[IP_MAX];
	int			port;
	int			packet_t;
	int			packet_len;
	void			*delimiter;
	int			delimiter_len;

	/* global log pointer */
        LOG     *log;
	/* Timer */
        TIMER *timer ;
	
	/* transaction  */
	CHUNK			*chunk;
	int			transaction_state;
	unsigned long		transaction_id;
	
	/* bytes total  */
	uint64_t 			send_total;
	uint64_t 			recv_total;
	uint64_t 			oob_recv_total;
	uint64_t 			oob_send_total;

	/* buffer setting */
	struct _BUFFER		*buffer; /* point to buffer */
	struct _BUFFER		*packet; /* point to packet */
	struct _BUFFER		*cache; /* point to cache */
	struct _BUFFER		*oob; /* point to OOB data */
	int			buf_size; /* max length of buffer */
	
	/* event setting */
	//struct event		s_event;
	//int			event_flags;
	
	EVENT *s_event;
	int			event_flags;
	
	
	/* queue setting */
	struct _QUEUE		*send_queue;
	struct _QUEUE		*joblist;

	int  (*set)(struct _SESSION *, int);
        void (*event_handler)(int, short, void *);
        int  (*event_update)(struct _SESSION *, short);
        int  (*read_handler)(struct _SESSION *);
        int  (*write_handler)(struct _SESSION *);
        void (*packet_handler)(struct _SESSION *);
        void (*data_handler)(struct _SESSION *);
	int  (*oob_handler)(struct _SESSION *);
	int  (*state_handler)(struct _SESSION *);
        void (*packet_reader)(struct _SESSION *);
	void (*push_message)(struct _SESSION *, int );
        int  (*push_cache)(struct _SESSION *, void *, size_t );
        int  (*push_chunk)(struct _SESSION *, void *, size_t );
        int  (*push_file)(struct _SESSION *, char *, uint64_t, uint64_t );
	int  (*chunk_reader)(struct _SESSION *);
        int  (*terminate)(struct _SESSION *);
        void (*clean)(struct _SESSION **);
}SESSION;
#endif


#ifndef _ERROR_LOG
#define _ERROR_LOG(format...){\
        fprintf(stderr, "[%s:%d] ", __FILE__, __LINE__);\
        fprintf(stderr, "\"");\
        fprintf(stderr, format);\
        fprintf(stderr, "\"");\
        fprintf(stderr, "\n");\
}
#endif

#ifndef _DEBUG_LOG
#ifdef  _DEBUG
#define _DEBUG_LOG(format...){\
        fprintf(stdout, "[%s:%d] ", __FILE__, __LINE__);\
        fprintf(stdout, "\"");\
        fprintf(stdout, format);\
        fprintf(stdout, "\"");\
        fprintf(stdout, "\n");\
}
#else
   #define _DEBUG_LOG(format...)
#endif
#endif


#ifdef __cplusplus
 }
#endif

#endif