gpfdist批量写入postgresql数据原理

int main(int argc, const char* const argv[])
{
if (gpfdist_init(argc, argv) == -1)
gfatal(NULL, "Initialization failed");
return gpfdist_run();
}

int gpfdist_run()
{
return event_dispatch();
}

int gpfdist_init(int argc, const char* const argv[])
{
/*初始化apr资源池*/
if (0 != apr_app_initialize(&argc, &argv, 0))
gfatal(NULL, "apr_app_initialize failed");
atexit(apr_terminate);
if (0 != apr_pool_create(&gcb.pool, 0))
gfatal(NULL, "apr_app_initialize failed");
//apr_signal_init(gcb.pool);
    gcb.session.tab = apr_hash_make(gcb.pool);
//解析命令行参数
parse_command_line(argc, argv, gcb.pool);
......
event_init();
signal_register();
//启动http服务
http_setup();
.....

http_setup(void)
{
SOCKET f;
int on = 1;
struct linger linger;
struct addrinfo hints;
struct addrinfo *addrs, *rp;
int  s;
int  i;
char service[32];
const char *hostaddr = NULL;
//绑定gpfdist的文件读写函数，用于从文件或其他方式读写数据
gpfdist_send    = gpfdist_socket_send;
gpfdist_receive = gpfdist_socket_receive;
......
/* 下面的内容就是从指定端口列表中测试哪些端口可用*/
for (;;)
{
//利用第一个端口组成socket使用的网络地址
        snprintf(service,32,"%d",opt.p);
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_UNSPEC;    /* Allow IPv4 or IPv6 */
hints.ai_socktype = SOCK_STREAM; /* tcp socket */
hints.ai_flags = AI_PASSIVE;    /* For wildcard IP address */
hints.ai_protocol = 0;          /* Any protocol */
s = getaddrinfo(hostaddr, service, &hints, &addrs);
.......
/* 测试地址是否可用，这个for循环只会执行一次，因为rp->ai_next=0*/
for (rp = addrs; rp != NULL; rp = rp->ai_next)
{
gprint(NULL, "Trying to open listening socket:\n");
print_listening_address(rp);
/*
* getaddrinfo gives us all the parameters for the socket() call
* as well as the parameters for the bind() call.
*/
f = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
//设置keep_alive linger等属性
             ......
if (bind(f, rp->ai_addr, rp->ai_addrlen) != 0)
{
......
}
/* listen with a big queue */
if (listen(f, opt.z))
{
......
}
gcb.listen_socks[gcb.listen_sock_count++] = f;
gprint(NULL, "Opening listening socket succeeded\n");
}
......
}
/*
* 为上述可用端口绑定可读事件响应函数do_accept，用于接收客户端的连接。
*/
for (i = 0; i < gcb.listen_sock_count; i++)
{
/* when this socket is ready, do accept */
event_set(&gcb.listen_events[i], gcb.listen_socks[i], EV_READ | EV_PERSIST,
do_accept, 0);
......
if (event_add(&gcb.listen_events[i], 0))
gfatal(NULL, "cannot set up event on listen socket: %s",
strerror(errno));
}
}

static void do_accept(int fd, short event, void* arg)
{
address_t           a;
socklen_t           len = sizeof(a);
SOCKET              sock;
request_t*          r;
apr_pool_t*         pool;
int                 on = 1;
struct linger       linger;
/* do the accept */
if ((sock = accept(fd, (struct sockaddr*) &a, &len)) < 0)
{
gwarning(NULL, "accept failed");
goto failure;
}
/* set to non-blocking, and close-on-exec */
......
/* set keepalive, reuseaddr, and linger */
......
/* create a pool container for this socket */
......
/* 调用setup_read为上述socket设置读事件响应函数do_read_request */
if (setup_read(r))
{
http_error(r, FDIST_INTERNAL_ERROR, "internal error");
request_end(r, 1, 0);
}
return;
}

create external table test
(
　　id integer,
　　name varchar
)
location (‘gpfdist://$IP:$PORT/$file_name’[,..])
format ‘csv’(delimiter’,’)
;