git.8kb.co.uk Git - pgpool-ii/pgpool-ii_2.2.5/blob - pcp/pcp.c

   1 /*
   2  * $Header: /cvsroot/pgpool/pgpool-II/pcp/pcp.c,v 1.8 2008/12/31 10:25:40 t-ishii Exp $
   3  *
   4  * Handles PCP connection, and protocol communication with pgpool-II
   5  * These are client APIs. Server program should use APIs in pcp_stream.c
   6  *
   7  *
   8  * pgpool: a language independent connection pool server for PostgreSQL
   9  * written by Tatsuo Ishii
  10  *
  11  * Copyright (c) 2003-2008      PgPool Global Development Group
  12  *
  13  * Permission to use, copy, modify, and distribute this software and
  14  * its documentation for any purpose and without fee is hereby
  15  * granted, provided that the above copyright notice appear in all
  16  * copies and that both that copyright notice and this permission
  17  * notice appear in supporting documentation, and that the name of the
  18  * author not be used in advertising or publicity pertaining to
  19  * distribution of the software without specific, written prior
  20  * permission. The author makes no representations about the
  21  * suitability of this software for any purpose.  It is provided "as
  22  * is" without express or implied warranty.
  23  *
  24  */
  25
  26 #include <stdio.h>
  27 #include <string.h>
  28 #include <stdlib.h>
  29 #include <sys/types.h>
  30 #include <sys/socket.h>
  31 #include <sys/time.h>
  32 #include <sys/un.h>
  33 #include <netinet/in.h>
  34 #include <netinet/tcp.h>
  35 #include <netdb.h>
  36 #include <unistd.h>
  37
  38 #include "pcp.h"
  39 #include "pcp_stream.h"
  40 #include "md5.h"
  41
  42 struct timeval pcp_timeout;
  43
  44 static PCP_CONNECTION *pc;
  45 #ifdef DEBUG
  46 static int debug = 1;
  47 #else
  48 static int debug = 0;
  49 #endif
  50 static int pcp_authorize(char *username, char *password);
  51
  52 /* --------------------------------
  53  * pcp_connect - open connection to pgpool using given arguments
  54  *
  55  * return 0 on success, -1 otherwise
  56  * --------------------------------
  57  */
  58 int
  59 pcp_connect(char *hostname, int port, char *username, char *password)
  60 {
  61         struct sockaddr_in addr;
  62         struct sockaddr_un unix_addr;
  63         struct hostent *hp;
  64         int fd;
  65         int on = 1;
  66         int len;
  67
  68         if (pc != NULL)
  69         {
  70                 if (debug) fprintf(stderr, "DEBUG: connection to backend \"%s\" already exists\n", hostname);
  71                 return 0;
  72         }
  73
  74         if (hostname == NULL || *hostname == '\0' || *hostname == '/')
  75         {
  76                 char *path;
  77
  78                 fd = socket(AF_UNIX, SOCK_STREAM, 0);
  79
  80                 if (fd < 0)
  81                 {
  82                         if (debug) fprintf(stderr, "DEBUG: could not create socket\n");
  83                         errorcode = SOCKERR;
  84                         return -1;
  85                 }
  86
  87                 memset(&unix_addr, 0, sizeof(unix_addr));
  88                 unix_addr.sun_family = AF_UNIX;
  89
  90                 if (hostname == NULL || *hostname == '\0')
  91                 {
  92                         path = UNIX_DOMAIN_PATH;
  93                 }
  94                 else
  95                 {
  96                         path = hostname;
  97                 }
  98
  99                 snprintf(unix_addr.sun_path, sizeof(unix_addr.sun_path), "%s/.s.PGSQL.%d",
 100                                  path, port);
 101
 102                 if (connect(fd, (struct sockaddr *) &unix_addr, sizeof(unix_addr)) < 0)
 103                 {
 104                         if (debug) fprintf(stderr, "DEBUG: could not connect to \"%s\"\n", unix_addr.sun_path);
 105                         close(fd);
 106                         errorcode = CONNERR;
 107                         return -1;
 108                 }
 109         }
 110         else
 111         {
 112                 fd = socket(AF_INET, SOCK_STREAM, 0);
 113                 if (fd < 0)
 114                 {
 115                         if (debug) fprintf(stderr, "DEBUG: could not create socket\n");
 116                         errorcode = SOCKERR;
 117                         return -1;
 118                 }
 119
 120                 if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
 121                                            (char *) &on, sizeof(on)) < 0)
 122                 {
 123                         if (debug) fprintf(stderr, "DEBUG: could not set socket option\n");
 124                         close(fd);
 125                         errorcode = SOCKERR;
 126                         return -1;
 127                 }
 128
 129                 memset((char *) &addr, 0, sizeof(addr));
 130                 ((struct sockaddr *) &addr)->sa_family = AF_INET;
 131                 hp = gethostbyname(hostname);
 132                 if ((hp == NULL) || (hp->h_addrtype != AF_INET))
 133                 {
 134                         if (debug) fprintf(stderr, "DEBUG: could not retrieve hostname\n");
 135                         close(fd);
 136                         errorcode = HOSTERR;
 137                         return -1;
 138                 }
 139                 memmove((char *) &(addr.sin_addr),
 140                                 (char *) hp->h_addr,
 141                                 hp->h_length);
 142                 addr.sin_port = htons(port);
 143
 144                 len = sizeof(struct sockaddr_in);
 145                 if (connect(fd, (struct sockaddr *) &addr, len) < 0)
 146                 {
 147                         if (debug) fprintf(stderr, "DEBUG: could not connect to \"%s\"\n", hostname);
 148                         close(fd);
 149                         errorcode = CONNERR;
 150                         return -1;
 151                 }
 152         }
 153
 154         pc = pcp_open(fd);
 155         if (pc == NULL)
 156         {
 157                 if (debug) fprintf(stderr, "DEBUG: could not allocate buffer space\n");
 158                 close(fd);
 159                 return -1;
 160         }
 161
 162         if (pcp_authorize(username, password) < 0)
 163         {
 164                 pcp_close(pc);
 165                 return -1;
 166         }
 167
 168         return 0;
 169 }
 170
 171 /* --------------------------------
 172  * pcp_authorize - authenticate with pgpool using username and password
 173  *
 174  * return 0 on success, -1 otherwise
 175  * --------------------------------
 176  */
 177 static int
 178 pcp_authorize(char *username, char *password)
 179 {
 180         char tos;
 181         char *buf = NULL;
 182         int wsize;
 183         int rsize;
 184         char salt[4];
 185         char encrypt_buf[(MD5_PASSWD_LEN+1)*2];
 186         char md5[MD5_PASSWD_LEN+1];
 187
 188         /* request salt */
 189         pcp_write(pc, "M", 1);
 190         wsize = htonl(sizeof(int));
 191         pcp_write(pc, &wsize, sizeof(int));
 192         if (pcp_flush(pc) < 0)
 193         {
 194                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 195                 return -1;
 196         }
 197
 198         if (pcp_read(pc, &tos, 1))
 199                 return -1;
 200         if (pcp_read(pc, &rsize, sizeof(int)))
 201                 return -1;
 202         rsize = ntohl(rsize);
 203         buf = (char *)malloc(rsize);
 204         if (buf == NULL)
 205         {
 206                 errorcode = NOMEMERR;
 207                 return -1;
 208         }
 209         if (pcp_read(pc, buf, rsize - sizeof(int)))
 210                 return -1;
 211         memcpy(salt, buf, 4);
 212         free(buf);
 213
 214         /* encrypt password */
 215         pool_md5_hash(password, strlen(password), md5);
 216         md5[MD5_PASSWD_LEN] = '\0';
 217
 218         pool_md5_encrypt(md5, username, strlen(username),
 219                                          encrypt_buf + MD5_PASSWD_LEN + 1);
 220         encrypt_buf[(MD5_PASSWD_LEN+1)*2-1] = '\0';
 221
 222         pool_md5_encrypt(encrypt_buf+MD5_PASSWD_LEN+1, salt, 4,
 223                                          encrypt_buf);
 224         encrypt_buf[MD5_PASSWD_LEN] = '\0';
 225
 226         pcp_write(pc, "R", 1);
 227         wsize = htonl((strlen(username)+1 + strlen(encrypt_buf)+1) + sizeof(int));
 228         pcp_write(pc, &wsize, sizeof(int));
 229         pcp_write(pc, username, strlen(username)+1);
 230         pcp_write(pc, encrypt_buf, strlen(encrypt_buf)+1);
 231         if (pcp_flush(pc) < 0)
 232         {
 233                 if  (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 234                 return -1;
 235         }
 236         if (debug) fprintf(stderr, "DEBUG: send: tos=\"R\", len=%d\n", ntohl(wsize));
 237
 238         if (pcp_read(pc, &tos, 1))
 239                 return -1;
 240         if (pcp_read(pc, &rsize, sizeof(int)))
 241                 return -1;
 242         rsize = ntohl(rsize);
 243         buf = (char *)malloc(rsize);
 244         if (buf == NULL)
 245         {
 246                 errorcode = NOMEMERR;
 247                 return -1;
 248         }
 249         if (pcp_read(pc, buf, rsize - sizeof(int)))
 250                 return -1;
 251         if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
 252
 253         if (tos == 'e')
 254         {
 255                 if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
 256                 errorcode = BACKENDERR;
 257         }
 258         else if (tos == 'r')
 259         {
 260                 if (strcmp(buf, "AuthenticationOK") == 0)
 261                 {
 262                         free(buf);
 263                         return 0;
 264                 }
 265
 266                 if (debug) fprintf(stderr, "DEBUG: authentication failed. reason=%s\n", buf);
 267                 errorcode = AUTHERR;
 268         }
 269         free(buf);
 270
 271         return -1;
 272 }
 273
 274 /* --------------------------------
 275  * pcp_disconnect - close connection to pgpool
 276  * --------------------------------
 277  */
 278 void
 279 pcp_disconnect(void)
 280 {
 281         int wsize;
 282
 283         if (pc == NULL)
 284         {
 285                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
 286                 return;
 287         }
 288
 289         pcp_write(pc, "X", 1);
 290         wsize = htonl(sizeof(int));
 291         pcp_write(pc, &wsize, sizeof(int));
 292         if (pcp_flush(pc) < 0)
 293         {
 294                 /* backend had closed connection already */
 295         }
 296         if (debug) fprintf(stderr, "DEBUG: send: tos=\"X\", len=%d\n", sizeof(int));
 297
 298         pcp_close(pc);
 299         pc = NULL;
 300 }
 301
 302 /* --------------------------------
 303  * pcp_terminate_pgpool - send terminate packet
 304  *
 305  * return 0 on success, -1 otherwise
 306  * --------------------------------
 307  */
 308 int
 309 pcp_terminate_pgpool(char mode)
 310 {
 311         int wsize;
 312
 313         if (pc == NULL)
 314         {
 315                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
 316                 errorcode = NOCONNERR;
 317                 return -1;
 318         }
 319
 320         pcp_write(pc, "T", 1);
 321         wsize = htonl(sizeof(int) + sizeof(char));
 322         pcp_write(pc, &wsize, sizeof(int));
 323         pcp_write(pc, &mode, sizeof(char));
 324         if (pcp_flush(pc) < 0)
 325         {
 326                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 327                 return -1;
 328         }
 329         if (debug) fprintf(stderr, "DEBUG: send: tos=\"T\", len=%d\n", ntohl(wsize));
 330
 331         return 0;
 332 }
 333
 334 /* --------------------------------
 335  * pcp_node_count - get number of nodes currently connected to pgpool
 336  *
 337  * return array of node IDs on success, -1 otherwise
 338  * --------------------------------
 339  */
 340 int
 341 pcp_node_count(void)
 342 {
 343         char tos;
 344         char *buf = NULL;
 345         int wsize;
 346         int rsize;
 347         char *index = NULL;
 348
 349         if (pc == NULL)
 350         {
 351                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
 352                 errorcode = NOCONNERR;
 353                 return -1;
 354         }
 355
 356         pcp_write(pc, "L", 1);
 357         wsize = htonl(sizeof(int));
 358         pcp_write(pc, &wsize, sizeof(int));
 359         if (pcp_flush(pc) < 0)
 360         {
 361                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 362                 return -1;
 363         }
 364         if (debug) fprintf(stderr, "DEBUG: send: tos=\"L\", len=%d\n", ntohl(wsize));
 365
 366         if (pcp_read(pc, &tos, 1))
 367                 return -1;
 368         if (pcp_read(pc, &rsize, sizeof(int)))
 369                 return -1;
 370         rsize = ntohl(rsize);
 371         buf = (char *)malloc(rsize);
 372         if (buf == NULL)
 373         {
 374                 errorcode = NOMEMERR;
 375                 return -1;
 376         }
 377         if (pcp_read(pc, buf, rsize - sizeof(int)))
 378         {
 379                 free(buf);
 380                 return -1;
 381         }
 382
 383         if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
 384
 385         if (tos == 'e')
 386         {
 387                 if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
 388                 errorcode = BACKENDERR;
 389         }
 390         else if (tos == 'l')
 391         {
 392                 if (strcmp(buf, "CommandComplete") == 0)
 393                 {
 394                         index = (char *) memchr(buf, '\0', rsize) + 1;
 395                         if (index != NULL)
 396                         {
 397                                 int ret = atoi(index);
 398                                 free(buf);
 399                                 return ret;
 400                         }
 401                 }
 402         }
 403
 404         free(buf);
 405
 406         return -1;
 407 }
 408
 409 /* --------------------------------
 410  * pcp_node_info - get information of node pointed by given argument
 411  *
 412  * return structure of node information on success, -1 otherwise
 413  * --------------------------------
 414  */
 415 BackendInfo *
 416 pcp_node_info(int nid)
 417 {
 418         int wsize;
 419         char node_id[16];
 420         char tos;
 421         char *buf = NULL;
 422         int rsize;
 423
 424         if (pc == NULL)
 425         {
 426                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
 427                 errorcode = NOCONNERR;
 428                 return NULL;
 429         }
 430
 431         snprintf(node_id, sizeof(node_id), "%d", nid);
 432
 433         pcp_write(pc, "I", 1);
 434         wsize = htonl(strlen(node_id)+1 + sizeof(int));
 435         pcp_write(pc, &wsize, sizeof(int));
 436         pcp_write(pc, node_id, strlen(node_id)+1);
 437         if (pcp_flush(pc) < 0)
 438         {
 439                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 440                 return NULL;
 441         }
 442         if (debug) fprintf(stderr, "DEBUG: send: tos=\"I\", len=%d\n", ntohl(wsize));
 443
 444         if (pcp_read(pc, &tos, 1))
 445                 return NULL;
 446         if (pcp_read(pc, &rsize, sizeof(int)))
 447                 return NULL;
 448         rsize = ntohl(rsize);
 449         buf = (char *)malloc(rsize);
 450         if (buf == NULL)
 451         {
 452                 errorcode = NOMEMERR;
 453                 return NULL;
 454         }
 455         if (pcp_read(pc, buf, rsize - sizeof(int)))
 456         {
 457                 free(buf);
 458                 return NULL;
 459         }
 460
 461         if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
 462
 463         if (tos == 'e')
 464         {
 465                 if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
 466                 errorcode = BACKENDERR;
 467                 free(buf);
 468                 return NULL;
 469         }
 470         else if (tos == 'i')
 471         {
 472                 if (strcmp(buf, "CommandComplete") == 0)
 473                 {
 474                         char *index = NULL;
 475                         BackendInfo* backend_info = NULL;
 476
 477                         backend_info = (BackendInfo *)malloc(sizeof(BackendInfo));
 478                         if (backend_info == NULL)
 479                         {
 480                                 errorcode = NOMEMERR;
 481                                 free(buf);
 482                                 return NULL;
 483                         }
 484                         // FIXME
 485 //                      rsize -= strlen("CommandComplete") + 1;
 486
 487                         index = (char *) memchr(buf, '\0', rsize) + 1;
 488                         if (index != NULL)
 489                                 strcpy(backend_info->backend_hostname, index);
 490
 491                         index = (char *) memchr(index, '\0', rsize) + 1;
 492                         if (index != NULL)
 493                                 backend_info->backend_port = atoi(index);
 494
 495                         index = (char *) memchr(index, '\0', rsize) + 1;
 496                         if (index != NULL)
 497                                 backend_info->backend_status = atoi(index);
 498
 499                         index = (char *) memchr(index, '\0', rsize) + 1;
 500                         if (index != NULL)
 501                                 backend_info->backend_weight = atof(index);
 502
 503                         free(buf);
 504                         return backend_info;
 505                 }
 506         }
 507
 508         free(buf);
 509         return NULL;
 510 }
 511
 512 /* --------------------------------
 513  * pcp_node_count - get number of nodes currently connected to pgpool
 514  *
 515  * return array of pids on success, NULL otherwise
 516  * --------------------------------
 517  */
 518 int *
 519 pcp_process_count(int *pnum)
 520 {
 521         char tos;
 522         char *buf = NULL;
 523         int wsize;
 524         int rsize;
 525
 526         if (pc == NULL)
 527         {
 528                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
 529                 errorcode = NOCONNERR;
 530                 return NULL;
 531         }
 532
 533         pcp_write(pc, "N", 1);
 534         wsize = htonl(sizeof(int));
 535         pcp_write(pc, &wsize, sizeof(int));
 536         if (pcp_flush(pc) < 0)
 537         {
 538                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 539                 return NULL;
 540         }
 541         if (debug) fprintf(stderr, "DEBUG: send: tos=\"N\", len=%d\n", ntohl(wsize));
 542
 543         if (pcp_read(pc, &tos, 1))
 544                 return NULL;
 545         if (pcp_read(pc, &rsize, sizeof(int)))
 546                 return NULL;
 547         rsize = ntohl(rsize);
 548         buf = (char *)malloc(rsize);
 549         if (buf == NULL)
 550         {
 551                 errorcode = NOMEMERR;
 552                 return NULL;
 553         }
 554         if (pcp_read(pc, buf, rsize - sizeof(int)))
 555         {
 556                 free(buf);
 557                 return NULL;
 558         }
 559         if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
 560
 561         if (tos == 'e')
 562         {
 563                 if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
 564                 free(buf);
 565                 errorcode = BACKENDERR;
 566                 return NULL;
 567         }
 568         else if (tos == 'n')
 569         {
 570                 if (strcmp(buf, "CommandComplete") == 0)
 571                 {
 572                         int process_count;
 573                         int *process_list = NULL;
 574                         char *index = NULL;
 575                         int i;
 576
 577                         index = (char *) memchr(buf, '\0', rsize) + 1;
 578                         process_count = atoi(index);
 579
 580                         process_list = (int *)malloc(sizeof(int) * process_count);
 581                         if (process_list == NULL)
 582                         {
 583                                 free(buf);
 584                                 errorcode = NOMEMERR;
 585                                 return NULL;
 586                         }
 587
 588                         for (i = 0; i < process_count; i++)
 589                         {
 590                                 index = (char *) memchr(index, '\0', rsize) + 1;
 591                                 process_list[i] = atoi(index);
 592                         }
 593
 594                         *pnum = process_count;
 595                         free(buf);
 596                         return process_list;
 597                 }
 598         }
 599
 600         free(buf);
 601         return NULL;
 602 }
 603
 604 /* --------------------------------
 605  * pcp_process_info - get information of node pointed by given argument
 606  *
 607  * return structure of process information on success, -1 otherwise
 608  * --------------------------------
 609  */
 610 ProcessInfo *
 611 pcp_process_info(int pid, int *array_size)
 612 {
 613         int wsize;
 614         char process_id[16];
 615         char tos;
 616         char *buf = NULL;
 617         int rsize;
 618
 619         ProcessInfo *process_info = NULL;
 620         int ci_size = 0;
 621         int offset = 0;
 622
 623         if (pc == NULL)
 624         {
 625                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
 626                 errorcode = NOCONNERR;
 627                 return NULL;
 628         }
 629
 630         snprintf(process_id, sizeof(process_id), "%d", pid);
 631
 632         pcp_write(pc, "P", 1);
 633         wsize = htonl(strlen(process_id)+1 + sizeof(int));
 634         pcp_write(pc, &wsize, sizeof(int));
 635         pcp_write(pc, process_id, strlen(process_id)+1);
 636         if (pcp_flush(pc) < 0)
 637         {
 638                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 639                 return NULL;
 640         }
 641         if (debug) fprintf(stderr, "DEBUG: send: tos=\"P\", len=%d\n", ntohl(wsize));
 642
 643         while (1)
 644         {
 645                 if (pcp_read(pc, &tos, 1))
 646                         return NULL;
 647                 if (pcp_read(pc, &rsize, sizeof(int)))
 648                         return NULL;
 649                 rsize = ntohl(rsize);
 650                 buf = (char *)malloc(rsize);
 651                 if (buf == NULL)
 652                 {
 653                         errorcode = NOMEMERR;
 654                         return NULL;
 655                 }
 656                 if (pcp_read(pc, buf, rsize - sizeof(int)))
 657                 {
 658                         free(buf);
 659                         return NULL;
 660                 }
 661                 if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
 662
 663                 if (tos == 'e')
 664                 {
 665                         if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
 666                         free(buf);
 667                         errorcode = BACKENDERR;
 668                         return NULL;
 669                 }
 670                 else if (tos == 'p')
 671                 {
 672                         char *index;
 673
 674                         if (strcmp(buf, "ArraySize") == 0)
 675                         {
 676                                 index = (char *) memchr(buf, '\0', rsize) + 1;
 677                                 if (index != NULL)
 678                                         ci_size = atoi(index);
 679
 680                                 *array_size = ci_size;
 681
 682                                 process_info = (ProcessInfo *)malloc(sizeof(ProcessInfo));
 683                                 if (process_info == NULL)
 684                                 {
 685                                         free(buf);
 686                                         errorcode = NOMEMERR;
 687                                         return NULL;
 688                                 }
 689                                 process_info->connection_info = NULL;
 690                                 process_info->connection_info = (ConnectionInfo *)malloc(sizeof(ConnectionInfo)*ci_size);
 691                                 if (process_info->connection_info == NULL)
 692                                 {
 693                                         free(buf);
 694                                         errorcode = NOMEMERR;
 695                                         return NULL;
 696                                 }
 697
 698                                 continue;
 699                         }
 700                         else if (strcmp(buf, "ProcessInfo") == 0)
 701                         {
 702                                 index = (char *) memchr(buf, '\0', rsize) + 1;
 703                                 if (index != NULL)
 704                                         strcpy(process_info->connection_info[offset].database, index);
 705
 706                                 index = (char *) memchr(index, '\0', rsize) + 1;
 707                                 if (index != NULL)
 708                                         strcpy(process_info->connection_info[offset].user, index);
 709
 710                                 index = (char *) memchr(index, '\0', rsize) + 1;
 711                                 if (index != NULL)
 712                                         process_info->start_time = atol(index);
 713
 714                                 index = (char *) memchr(index, '\0', rsize) + 1;
 715                                 if (index != NULL)
 716                                         process_info->connection_info[offset].create_time = atol(index);
 717
 718                                 index = (char *) memchr(index, '\0', rsize) + 1;
 719                                 if (index != NULL)
 720                                         process_info->connection_info[offset].major = atoi(index);
 721
 722                                 index = (char *) memchr(index, '\0', rsize) + 1;
 723                                 if (index != NULL)
 724                                         process_info->connection_info[offset].minor = atoi(index);
 725
 726                                 index = (char *) memchr(index, '\0', rsize) + 1;
 727                                 if (index != NULL)
 728                                         process_info->connection_info[offset].counter = atoi(index);
 729
 730                                 offset++;
 731                         }
 732                         else if (strcmp(buf, "CommandComplete") == 0)
 733                         {
 734                                 free(buf);
 735                                 return process_info;
 736                         }
 737                         else
 738                         {
 739                                 // never reached
 740                         }
 741                 }
 742         }
 743
 744         free(buf);
 745         return NULL;
 746 }
 747
 748 /* --------------------------------
 749  * pcp_systemdb_info - get information of system DB
 750  *
 751  * return structure of system DB information on success, -1 otherwise
 752  * --------------------------------
 753  */
 754 SystemDBInfo *
 755 pcp_systemdb_info(void)
 756 {
 757         char tos;
 758         char *buf = NULL;
 759         int wsize;
 760         int rsize;
 761         SystemDBInfo *systemdb_info = NULL;
 762         int offset = 0;
 763
 764         if (pc == NULL)
 765         {
 766                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
 767                 errorcode = NOCONNERR;
 768                 return NULL;
 769         }
 770
 771         pcp_write(pc, "S", 1);
 772         wsize = htonl(sizeof(int));
 773         pcp_write(pc, &wsize, sizeof(int));
 774         if (pcp_flush(pc) < 0)
 775         {
 776                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
 777                 return NULL;
 778         }
 779         if (debug) fprintf(stderr, "DEBUG: send: tos=\"S\", len=%d\n", ntohl(wsize));
 780
 781         while (1) {
 782                 if (pcp_read(pc, &tos, 1))
 783                         return NULL;
 784                 if (pcp_read(pc, &rsize, sizeof(int)))
 785                         return NULL;
 786                 rsize = ntohl(rsize);
 787                 buf = (char *)malloc(rsize);
 788                 if (buf == NULL)
 789                 {
 790                         errorcode = NOMEMERR;
 791                         return NULL;
 792                 }
 793                 if (pcp_read(pc, buf, rsize - sizeof(int)))
 794                 {
 795                         free(buf);
 796                         return NULL;
 797                 }
 798                 if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
 799
 800                 if (tos == 'e')
 801                 {
 802                         if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
 803                         free(buf);
 804                         errorcode = BACKENDERR;
 805                         return NULL;
 806                 }
 807                 else if (tos == 's')
 808                 {
 809                         char *index;
 810
 811                         if (strcmp(buf, "SystemDBInfo") == 0)
 812                         {
 813                                 systemdb_info = (SystemDBInfo *)malloc(sizeof(SystemDBInfo));
 814                                 if (systemdb_info == NULL)
 815                                 {
 816                                         free(buf);
 817                                         errorcode = NOMEMERR;
 818                                         return NULL;
 819                                 }
 820
 821                                 index = (char *) memchr(buf, '\0', rsize) + 1;
 822                                 if (index != NULL)
 823                                         systemdb_info->hostname = strdup(index);
 824                                 if (systemdb_info->hostname == NULL)
 825                                 {
 826                                         free(buf);
 827                                         free_systemdb_info(systemdb_info);
 828                                         errorcode = NOMEMERR;
 829                                         return NULL;
 830                                 }
 831
 832                                 index = (char *) memchr(index, '\0', rsize) + 1;
 833                                 if (index != NULL)
 834                                         systemdb_info->port = atoi(index);
 835
 836                                 index = (char *) memchr(index, '\0', rsize) + 1;
 837                                 if (index != NULL)
 838                                         systemdb_info->user = strdup(index);
 839                                 if (systemdb_info->user == NULL)
 840                                 {
 841                                         free(buf);
 842                                         free_systemdb_info(systemdb_info);
 843                                         errorcode = NOMEMERR;
 844                                         return NULL;
 845                                 }
 846
 847                                 index = (char *) memchr(index, '\0', rsize) + 1;
 848                                 if (index != NULL)
 849                                         systemdb_info->password = strdup(index);
 850                                 if (systemdb_info->password == NULL)
 851                                 {
 852                                         free(buf);
 853                                         free_systemdb_info(systemdb_info);
 854                                         errorcode = NOMEMERR;
 855                                         return NULL;
 856                                 }
 857
 858                                 index = (char *) memchr(index, '\0', rsize) + 1;
 859                                 if (index != NULL)
 860                                         systemdb_info->schema_name = strdup(index);
 861                                 if (systemdb_info->schema_name == NULL)
 862                                 {
 863                                         free(buf);
 864                                         free_systemdb_info(systemdb_info);
 865                                         errorcode = NOMEMERR;
 866                                         return NULL;
 867                                 }
 868
 869                                 index = (char *) memchr(index, '\0', rsize) + 1;
 870                                 if (index != NULL)
 871                                         systemdb_info->database_name = strdup(index);
 872                                 if (systemdb_info->database_name == NULL)
 873                                 {
 874                                         free(buf);
 875                                         free_systemdb_info(systemdb_info);
 876                                         errorcode = NOMEMERR;
 877                                         return NULL;
 878                                 }
 879
 880                                 index = (char *) memchr(index, '\0', rsize) + 1;
 881                                 if (index != NULL)
 882                                         systemdb_info->dist_def_num = atoi(index);
 883
 884                                 index = (char *) memchr(index, '\0', rsize) + 1;
 885                                 if (index != NULL)
 886                                         systemdb_info->system_db_status = atoi(index);
 887
 888                                 if (systemdb_info->dist_def_num > 0)
 889                                 {
 890                                         systemdb_info->dist_def_slot = NULL;
 891                                         systemdb_info->dist_def_slot = (DistDefInfo *)malloc(sizeof(DistDefInfo) * systemdb_info->dist_def_num);
 892                                         if (systemdb_info->dist_def_slot == NULL)
 893                                         {
 894                                                 free(buf);
 895                                                 free_systemdb_info(systemdb_info);
 896                                                 errorcode = NOMEMERR;
 897                                                 return NULL;
 898                                         }
 899                                 }
 900                         }
 901                         else if (strcmp(buf, "DistDefInfo") == 0)
 902                         {
 903                                 DistDefInfo *dist_def_info = NULL;
 904                                 int i;
 905
 906                                 dist_def_info = (DistDefInfo *)malloc(sizeof(DistDefInfo));
 907                                 if (dist_def_info == NULL)
 908                                 {
 909                                         free(buf);
 910                                         free_systemdb_info(systemdb_info);
 911                                         errorcode = NOMEMERR;
 912                                         return NULL;
 913                                 }
 914
 915                                 index = (char *) memchr(buf, '\0', rsize) + 1;
 916                                 if (index != NULL)
 917                                         dist_def_info->dbname = strdup(index);
 918                                 if (dist_def_info->dbname == NULL)
 919                                 {
 920                                         free(buf);
 921                                         free_systemdb_info(systemdb_info);
 922                                         errorcode = NOMEMERR;
 923                                         return NULL;
 924                                 }
 925
 926                                 index = (char *) memchr(index, '\0', rsize) + 1;
 927                                 if (index != NULL)
 928                                         dist_def_info->schema_name = strdup(index);
 929                                 if (dist_def_info->schema_name == NULL)
 930                                 {
 931                                         free(buf);
 932                                         free_systemdb_info(systemdb_info);
 933                                         errorcode = NOMEMERR;
 934                                         return NULL;
 935                                 }
 936
 937                                 index = (char *) memchr(index, '\0', rsize) + 1;
 938                                 if (index != NULL)
 939                                         dist_def_info->table_name = strdup(index);
 940                                 if (dist_def_info->table_name == NULL)
 941                                 {
 942                                         free(buf);
 943                                         free_systemdb_info(systemdb_info);
 944                                         errorcode = NOMEMERR;
 945                                         return NULL;
 946                                 }
 947
 948                                 index = (char *) memchr(index, '\0', rsize) + 1;
 949                                 if (index != NULL)
 950                                         dist_def_info->dist_key_col_name = strdup(index);
 951                                 if (dist_def_info->dist_key_col_name == NULL)
 952                                 {
 953                                         free(buf);
 954                                         free_systemdb_info(systemdb_info);
 955                                         errorcode = NOMEMERR;
 956                                         return NULL;
 957                                 }
 958
 959                                 index = (char *) memchr(index, '\0', rsize) + 1;
 960                                 if (index != NULL)
 961                                         dist_def_info->col_num = atoi(index);
 962
 963                                 dist_def_info->col_list = NULL;
 964                                 dist_def_info->col_list = (char **)malloc(sizeof(char *) * dist_def_info->col_num);
 965                                 if (dist_def_info->col_list == NULL)
 966                                 {
 967                                         free(buf);
 968                                         free_systemdb_info(systemdb_info);
 969                                         errorcode = NOMEMERR;
 970                                         return NULL;
 971                                 }
 972                                 for (i = 0; i < dist_def_info->col_num; i++)
 973                                 {
 974                                         index = (char *) memchr(index, '\0', rsize) + 1;
 975                                         if (index != NULL)
 976                                                 dist_def_info->col_list[i] = strdup(index);
 977                                         if (dist_def_info->col_list[i] == NULL)
 978                                         {
 979                                                 free(buf);
 980                                                 free_systemdb_info(systemdb_info);
 981                                                 errorcode = NOMEMERR;
 982                                                 return NULL;
 983                                         }
 984                                 }
 985
 986                                 dist_def_info->type_list = NULL;
 987                                 dist_def_info->type_list = (char **)malloc(sizeof(char *) * dist_def_info->col_num);
 988                                 if (dist_def_info->type_list == NULL)
 989                                 {
 990                                         free(buf);
 991                                         free_systemdb_info(systemdb_info);
 992                                         errorcode = NOMEMERR;
 993                                         return NULL;
 994                                 }
 995                                 for (i = 0; i < dist_def_info->col_num; i++)
 996                                 {
 997                                         index = (char *) memchr(index, '\0', rsize) + 1;
 998                                         if (index != NULL)
 999                                                 dist_def_info->type_list[i] = strdup(index);
1000                                         if (dist_def_info->type_list[i] == NULL)
1001                                         {
1002                                                 free(buf);
1003                                                 free_systemdb_info(systemdb_info);
1004                                                 errorcode = NOMEMERR;
1005                                                 return NULL;
1006                                         }
1007                                 }
1008
1009                                 index = (char *) memchr(index, '\0', rsize) + 1;
1010                                 if (index != NULL)
1011                                         dist_def_info->dist_def_func = strdup(index);
1012                                 if (dist_def_info->dist_def_func == NULL)
1013                                 {
1014                                         free(buf);
1015                                         free_systemdb_info(systemdb_info);
1016                                         errorcode = NOMEMERR;
1017                                         return NULL;
1018                                 }
1019
1020                                 memcpy(&systemdb_info->dist_def_slot[offset++], dist_def_info, sizeof(DistDefInfo));
1021                         }
1022                         else if (strcmp(buf, "CommandComplete") == 0)
1023                         {
1024                                 free(buf);
1025                                 return systemdb_info;
1026                         }
1027                         else
1028                         {
1029                                 // never reached
1030                         }
1031                 }
1032         }
1033
1034         free(buf);
1035         return NULL;
1036 }
1037
1038 void
1039 free_systemdb_info(SystemDBInfo * si)
1040 {
1041         int i, j;
1042
1043         free(si->hostname);
1044         free(si->user);
1045         free(si->password);
1046         free(si->schema_name);
1047         free(si->database_name);
1048
1049         if (si->dist_def_slot != NULL)
1050         {
1051                 for (i = 0; i < si->dist_def_num; i++)
1052                 {
1053                         DistDefInfo *di = &si->dist_def_slot[i];
1054                         free(di->dbname);
1055                         free(di->schema_name);
1056                         free(di->table_name);
1057                         free(di->dist_def_func);
1058                         for (j = 0; j < di->col_num; j++)
1059                         {
1060                                 free(di->col_list[j]);
1061                                 free(di->type_list[j]);
1062                         }
1063                 }
1064         }
1065
1066         free(si);
1067 }
1068
1069 /* --------------------------------
1070  * pcp_detach_node - dettach a node given by the argument from pgpool's control
1071  *
1072  * return 0 on success, -1 otherwise
1073  * --------------------------------
1074  */
1075 int
1076 pcp_detach_node(int nid)
1077 {
1078         int wsize;
1079         char node_id[16];
1080         char tos;
1081         char *buf = NULL;
1082         int rsize;
1083
1084         if (pc == NULL)
1085         {
1086                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
1087                 errorcode = NOCONNERR;
1088                 return -1;
1089         }
1090
1091         snprintf(node_id, sizeof(node_id), "%d", nid);
1092
1093         pcp_write(pc, "D", 1);
1094         wsize = htonl(strlen(node_id)+1 + sizeof(int));
1095         pcp_write(pc, &wsize, sizeof(int));
1096         pcp_write(pc, node_id, strlen(node_id)+1);
1097         if (pcp_flush(pc) < 0)
1098         {
1099                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
1100                 return -1;
1101         }
1102         if (debug) fprintf(stderr, "DEBUG: send: tos=\"D\", len=%d\n", ntohl(wsize));
1103
1104         if (pcp_read(pc, &tos, 1))
1105                 return -1;
1106         if (pcp_read(pc, &rsize, sizeof(int)))
1107                 return -1;
1108         rsize = ntohl(rsize);
1109         buf = (char *)malloc(rsize);
1110         if (buf == NULL)
1111         {
1112                 errorcode = NOMEMERR;
1113                 return -1;
1114         }
1115         if (pcp_read(pc, buf, rsize - sizeof(int)))
1116         {
1117                 free(buf);
1118                 return -1;
1119         }
1120         if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
1121
1122         if (tos == 'e')
1123         {
1124                 if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
1125                 errorcode = BACKENDERR;
1126         }
1127         else if (tos == 'd')
1128         {
1129                 /* strcmp() for success message, or fail */
1130                 if(strcmp(buf, "CommandComplete") == 0)
1131                 {
1132                         free(buf);
1133                         return 0;
1134                 }
1135         }
1136
1137         free(buf);
1138         return -1;
1139 }
1140
1141
1142 /* --------------------------------
1143  * pcp_attach_node - attach a node given by the argument from pgpool's control
1144  *
1145  * return 0 on success, -1 otherwise
1146  * --------------------------------
1147  */
1148 int
1149 pcp_attach_node(int nid)
1150 {
1151         int wsize;
1152         char node_id[16];
1153         char tos;
1154         char *buf = NULL;
1155         int rsize;
1156
1157         if (pc == NULL)
1158         {
1159                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
1160                 errorcode = NOCONNERR;
1161                 return -1;
1162         }
1163
1164         snprintf(node_id, sizeof(node_id), "%d", nid);
1165
1166         pcp_write(pc, "C", 1);
1167         wsize = htonl(strlen(node_id)+1 + sizeof(int));
1168         pcp_write(pc, &wsize, sizeof(int));
1169         pcp_write(pc, node_id, strlen(node_id)+1);
1170         if (pcp_flush(pc) < 0)
1171         {
1172                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
1173                 return -1;
1174         }
1175         if (debug) fprintf(stderr, "DEBUG: send: tos=\"D\", len=%d\n", ntohl(wsize));
1176
1177         if (pcp_read(pc, &tos, 1))
1178                 return -1;
1179         if (pcp_read(pc, &rsize, sizeof(int)))
1180                 return -1;
1181         rsize = ntohl(rsize);
1182         buf = (char *)malloc(rsize);
1183         if (buf == NULL)
1184         {
1185                 errorcode = NOMEMERR;
1186                 return -1;
1187         }
1188         if (pcp_read(pc, buf, rsize - sizeof(int)))
1189         {
1190                 free(buf);
1191                 return -1;
1192         }
1193         if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
1194
1195         if (tos == 'e')
1196         {
1197                 if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
1198                 errorcode = BACKENDERR;
1199         }
1200         else if (tos == 'c')
1201         {
1202                 /* strcmp() for success message, or fail */
1203                 if(strcmp(buf, "CommandComplete") == 0)
1204                 {
1205                         free(buf);
1206                         return 0;
1207                 }
1208         }
1209
1210         free(buf);
1211         return -1;
1212 }
1213
1214 void
1215 pcp_set_timeout(long sec)
1216 {
1217         /* disable timeout (wait forever!) (2008/02/08 yamaguti) */
1218         sec = 0;
1219         pcp_timeout.tv_sec = sec;
1220 }
1221
1222
1223 int
1224 pcp_recovery_node(int nid)
1225 {
1226         int wsize;
1227         char node_id[16];
1228         char tos;
1229         char *buf = NULL;
1230         int rsize;
1231
1232         if (pc == NULL)
1233         {
1234                 if (debug) fprintf(stderr, "DEBUG: connection does not exist\n");
1235                 errorcode = NOCONNERR;
1236                 return -1;
1237         }
1238
1239         snprintf(node_id, sizeof(node_id), "%d", nid);
1240
1241         pcp_write(pc, "O", 1);
1242         wsize = htonl(strlen(node_id)+1 + sizeof(int));
1243         pcp_write(pc, &wsize, sizeof(int));
1244         pcp_write(pc, node_id, strlen(node_id)+1);
1245         if (pcp_flush(pc) < 0)
1246         {
1247                 if (debug) fprintf(stderr, "DEBUG: could not send data to backend\n");
1248                 return -1;
1249         }
1250         if (debug) fprintf(stderr, "DEBUG: send: tos=\"D\", len=%d\n", ntohl(wsize));
1251
1252         if (pcp_read(pc, &tos, 1))
1253                 return -1;
1254         if (pcp_read(pc, &rsize, sizeof(int)))
1255                 return -1;
1256         rsize = ntohl(rsize);
1257         buf = (char *)malloc(rsize);
1258         if (buf == NULL)
1259         {
1260                 errorcode = NOMEMERR;
1261                 return -1;
1262         }
1263         if (pcp_read(pc, buf, rsize - sizeof(int)))
1264         {
1265                 free(buf);
1266                 return -1;
1267         }
1268         if (debug) fprintf(stderr, "DEBUG: recv: tos=\"%c\", len=%d, data=%s\n", tos, rsize, buf);
1269
1270         if (tos == 'e')
1271         {
1272                 if (debug) fprintf(stderr, "DEBUG: command failed. reason=%s\n", buf);
1273                 errorcode = BACKENDERR;
1274         }
1275         else if (tos == 'c')
1276         {
1277                 /* strcmp() for success message, or fail */
1278                 if(strcmp(buf, "CommandComplete") == 0)
1279                 {
1280                         free(buf);
1281                         return 0;
1282                 }
1283         }
1284
1285         free(buf);
1286         return -1;
1287 }
1288
1289 void
1290 pcp_enable_debug(void)
1291 {
1292         debug = 1;
1293 }
1294
1295 void
1296 pcp_disable_debug(void)
1297 {
1298         debug = 0;
1299 }