src/yaz-pdu-assoc.cpp

   1 /*
   2  * Copyright (c) 1998-2000, Index Data.
   3  * See the file LICENSE for details.
   4  *
   5  * $Log: yaz-pdu-assoc.cpp,v $
   6  * Revision 1.20  2000-11-20 11:27:33  adam
   7  * Fixes for connect operation (timeout and notify fix).
   8  *
   9  * Revision 1.19  2000/11/01 14:22:59  adam
  10  * Added fd parameter for method IYaz_PDU_Observer::clone.
  11  *
  12  * Revision 1.18  2000/10/24 12:29:57  adam
  13  * Fixed bug in proxy where a Yaz_ProxyClient could be owned by
  14  * two Yaz_Proxy's (fatal).
  15  *
  16  * Revision 1.17  2000/10/11 11:58:16  adam
  17  * Moved header files to include/yaz++. Switched to libtool and automake.
  18  * Configure script creates yaz++-config script.
  19  *
  20  * Revision 1.16  2000/09/22 09:54:11  heikki
  21  * minor
  22  *
  23  * Revision 1.15  2000/09/21 21:43:20  adam
  24  * Better high-level server API.
  25  *
  26  * Revision 1.14  2000/09/12 12:09:53  adam
  27  * More work on high-level server.
  28  *
  29  * Revision 1.13  2000/09/08 10:23:42  adam
  30  * Added skeleton of yaz-z-server.
  31  *
  32  * Revision 1.12  2000/09/06 14:23:45  adam
  33  * WIN32 updates.
  34  *
  35  * Revision 1.11  2000/09/04 08:29:22  adam
  36  * Fixed memory leak(s). Added re-use of associations, rather than
  37  * re-init, when maximum number of targets are in use.
  38  *
  39  * Revision 1.10  2000/08/10 08:42:42  adam
  40  * Fixes for {set,get}_APDU_log.
  41  *
  42  * Revision 1.9  1999/12/06 13:52:45  adam
  43  * Modified for new location of YAZ header files. Experimental threaded
  44  * operation.
  45  *
  46  * Revision 1.8  1999/04/28 13:04:03  adam
  47  * Fixed setting of proxy otherInfo so that database(s) are removed.
  48  *
  49  * Revision 1.7  1999/04/21 12:09:01  adam
  50  * Many improvements. Modified to proxy server to work with "sessions"
  51  * based on cookies.
  52  *
  53  * Revision 1.6  1999/04/20 10:30:05  adam
  54  * Implemented various stuff for client and proxy. Updated calls
  55  * to ODR to reflect new name parameter.
  56  *g
  57  * Revision 1.5  1999/04/09 11:46:57  adam
  58  * Added object Yaz_Z_Assoc. Much more functional client.
  59  *
  60  * Revision 1.4  1999/03/23 14:17:57  adam
  61  * More work on timeout handling. Work on yaz-client.
  62  *
  63  * Revision 1.3  1999/02/02 14:01:20  adam
  64  * First WIN32 port of YAZ++.
  65  *
  66  * Revision 1.2  1999/01/28 13:08:44  adam
  67  * Yaz_PDU_Assoc better encapsulated. Memory leak fix in
  68  * yaz-socket-manager.cc.
  69  *
  70  * Revision 1.1.1.1  1999/01/28 09:41:07  adam
  71  * First implementation of YAZ++.
  72  *
  73  */
  74
  75 #include <assert.h>
  76
  77 #include <yaz++/yaz-pdu-assoc.h>
  78
  79 #include <yaz/log.h>
  80 #include <yaz/tcpip.h>
  81
  82 Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable)
  83 {
  84     m_state = Closed;
  85     m_cs = 0;
  86     m_socketObservable = socketObservable;
  87     m_PDU_Observer = 0;
  88     m_queue_out = 0;
  89     m_input_buf = 0;
  90     m_input_len = 0;
  91     m_children = 0;
  92     m_parent = 0;
  93     m_next = 0;
  94     m_destroyed = 0;
  95     m_idleTime = 0;
  96     m_log = LOG_DEBUG;
  97 }
  98
  99 IYaz_PDU_Observable *Yaz_PDU_Assoc::clone()
 100 {
 101     Yaz_PDU_Assoc *copy = new Yaz_PDU_Assoc(m_socketObservable);
 102     return copy;
 103 }
 104
 105 void Yaz_PDU_Assoc::socketNotify(int event)
 106 {
 107     logf (m_log, "Yaz_PDU_Assoc::socketNotify p=%p event = %d", this, event);
 108     switch (m_state)
 109     {
 110     case Connecting:
 111         if (event & YAZ_SOCKET_OBSERVE_READ)
 112         {
 113             close();
 114             m_PDU_Observer->failNotify();
 115         }
 116         else if (event & YAZ_SOCKET_OBSERVE_TIMEOUT)
 117         {
 118             m_PDU_Observer->timeoutNotify();
 119         }
 120         else
 121         {
 122             m_state = Ready;
 123             m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 124                                              YAZ_SOCKET_OBSERVE_EXCEPT);
 125             m_PDU_Observer->connectNotify();
 126             flush_PDU();
 127         }
 128         break;
 129     case Listen:
 130         if (event & YAZ_SOCKET_OBSERVE_READ)
 131         {
 132             int res;
 133             COMSTACK new_line;
 134
 135             if ((res = cs_listen(m_cs, 0, 0)) == 1)
 136                 return;
 137             if (res < 0)
 138             {
 139                 logf(LOG_FATAL|LOG_ERRNO, "cs_listen failed");
 140                 return;
 141             }
 142             if (!(new_line = cs_accept(m_cs)))
 143                 return;
 144             /* 1. create socket-manager
 145                2. create pdu-assoc
 146                3. create top-level object
 147                     setup observer for child fileid in pdu-assoc
 148                4. start thread
 149             */
 150             int fd = cs_fileno(new_line);
 151             logf (m_log, "accept ok fd = %d", fd);
 152             cs_fileno(new_line) = -1;
 153             cs_close (new_line);
 154             childNotify(fd);
 155         }
 156         break;
 157     case Ready:
 158         if (event & YAZ_SOCKET_OBSERVE_WRITE)
 159         {
 160             flush_PDU();
 161         }
 162         if (event & YAZ_SOCKET_OBSERVE_READ)
 163         {
 164             do
 165             {
 166                 int res = cs_get (m_cs, &m_input_buf, &m_input_len);
 167                 if (res == 1)
 168                     return;
 169                 else if (res <= 0)
 170                 {
 171                     logf (m_log, "Connection closed by peer");
 172                     close();
 173                     m_PDU_Observer->failNotify(); // problem here..
 174                     return;
 175                 }
 176                 // lock it, so we know if recv_PDU deletes it.
 177                 int destroyed = 0;
 178                 m_destroyed = &destroyed;
 179
 180                 m_PDU_Observer->recv_PDU(m_input_buf, res);
 181                 m_destroyed = 0;
 182                 if (destroyed)   // it really was destroyed, return now.
 183                     return;
 184             } while (m_cs && cs_more (m_cs));
 185         }
 186         if (event & YAZ_SOCKET_OBSERVE_TIMEOUT)
 187         {
 188             m_PDU_Observer->timeoutNotify();
 189         }
 190         break;
 191     case Closed:
 192         logf (m_log, "CLOSING state=%d event was %d", m_state, event);
 193         close();
 194         m_PDU_Observer->failNotify();
 195         break;
 196     default:
 197         logf (m_log, "Unknown state=%d event was %d", m_state, event);
 198         close();
 199         m_PDU_Observer->failNotify();
 200     }
 201 }
 202
 203 void Yaz_PDU_Assoc::close()
 204 {
 205     m_socketObservable->deleteObserver(this);
 206     m_state = Closed;
 207     if (m_cs)
 208     {
 209         logf (m_log, "Yaz_PDU_Assoc::close fd=%d", cs_fileno(m_cs));
 210         cs_close (m_cs);
 211     }
 212     m_cs = 0;
 213     while (m_queue_out)
 214     {
 215         PDU_Queue *q_this = m_queue_out;
 216         m_queue_out = m_queue_out->m_next;
 217         delete q_this;
 218     }
 219     xfree (m_input_buf);
 220     m_input_buf = 0;
 221     m_input_len = 0;
 222 }
 223
 224 void Yaz_PDU_Assoc::destroy()
 225 {
 226     close();
 227     if (m_destroyed)
 228         *m_destroyed = 1;
 229     Yaz_PDU_Assoc **c;
 230
 231     // delete from parent's child list (if any)
 232     if (m_parent)
 233     {
 234         c = &m_parent->m_children;
 235         while (*c != this)
 236         {
 237             assert (*c);
 238             c = &(*c)->m_next;
 239         }
 240         *c = (*c)->m_next;
 241     }
 242     // delete all children ...
 243     c = &m_children;
 244     while (*c)
 245     {
 246         Yaz_PDU_Assoc *here = *c;
 247         *c = (*c)->m_next;
 248         here->m_parent = 0;
 249         delete here;
 250     }
 251 }
 252
 253 Yaz_PDU_Assoc::PDU_Queue::PDU_Queue(const char *buf, int len)
 254 {
 255     m_buf = (char *) malloc (len);
 256     memcpy (m_buf, buf, len);
 257     m_len = len;
 258     m_next = 0;
 259 }
 260
 261 Yaz_PDU_Assoc::PDU_Queue::~PDU_Queue()
 262 {
 263     free (m_buf);
 264 }
 265
 266 int Yaz_PDU_Assoc::flush_PDU()
 267 {
 268     int r;
 269
 270     logf (m_log, "Yaz_PDU_Assoc::flush_PDU");
 271     if (m_state != Ready)
 272     {
 273         logf (m_log, "YAZ_PDU_Assoc::flush_PDU, not ready");
 274         return 1;
 275     }
 276     PDU_Queue *q = m_queue_out;
 277     if (!q)
 278     {
 279         m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 280                                          YAZ_SOCKET_OBSERVE_EXCEPT);
 281         return 0;
 282     }
 283     r = cs_put (m_cs, q->m_buf, q->m_len);
 284     if (r < 0)
 285     {
 286         close();
 287         m_PDU_Observer->failNotify();
 288         return r;
 289     }
 290     if (r == 1)
 291     {
 292         m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 293                                          YAZ_SOCKET_OBSERVE_EXCEPT|
 294                                          YAZ_SOCKET_OBSERVE_WRITE);
 295         logf (m_log, "Yaz_PDU_Assoc::flush_PDU put %d bytes (incomplete)",
 296               q->m_len);
 297         return r;
 298     }
 299     logf (m_log, "Yaz_PDU_Assoc::flush_PDU put %d bytes", q->m_len);
 300     // whole packet sent... delete this and proceed to next ...
 301     m_queue_out = q->m_next;
 302     delete q;
 303     // don't select on write if queue is empty ...
 304     if (!m_queue_out)
 305         m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 306                                          YAZ_SOCKET_OBSERVE_EXCEPT);
 307     return r;
 308 }
 309
 310 int Yaz_PDU_Assoc::send_PDU(const char *buf, int len)
 311 {
 312     logf (m_log, "Yaz_PDU_Assoc::send_PDU");
 313     PDU_Queue **pq = &m_queue_out;
 314     int is_idle = (*pq ? 0 : 1);
 315
 316     if (!m_cs)
 317     {
 318         logf (m_log, "Yaz_PDU_Assoc::send_PDU failed, m_cs == 0");
 319         return -1;
 320     }
 321     while (*pq)
 322         pq = &(*pq)->m_next;
 323     *pq = new PDU_Queue(buf, len);
 324     if (is_idle)
 325         return flush_PDU ();
 326     else
 327         logf (m_log, "Yaz_PDU_Assoc::cannot send_PDU fd=%d",
 328               cs_fileno(m_cs));
 329     return 0;
 330 }
 331
 332 COMSTACK Yaz_PDU_Assoc::comstack()
 333 {
 334     if (!m_cs)
 335     {
 336         CS_TYPE cs_type = tcpip_type;
 337         m_cs = cs_create (cs_type, 0, PROTO_Z3950);
 338     }
 339     return m_cs;
 340 }
 341
 342 void Yaz_PDU_Assoc::listen(IYaz_PDU_Observer *observer,
 343                            const char *addr)
 344 {
 345     close();
 346     void *ap;
 347     COMSTACK cs = comstack();
 348
 349     logf (m_log, "Yaz_PDU_Assoc::listen %s", addr);
 350     m_PDU_Observer = observer;
 351     if (!cs)
 352         return;
 353     ap = cs_straddr (cs, addr);
 354     if (!ap)
 355         return;
 356     if (cs_bind(cs, ap, CS_SERVER) < 0)
 357         return;
 358     m_socketObservable->addObserver(cs_fileno(cs), this);
 359     m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 360                                      YAZ_SOCKET_OBSERVE_EXCEPT);
 361     logf (m_log, "Yaz_PDU_Assoc::listen ok fd=%d", cs_fileno(cs));
 362     m_state = Listen;
 363 }
 364
 365 void Yaz_PDU_Assoc::idleTime(int idleTime)
 366 {
 367     m_idleTime = idleTime;
 368     logf (m_log, "Yaz_PDU_Assoc::idleTime(%d)", idleTime);
 369     m_socketObservable->timeoutObserver(this, m_idleTime);
 370 }
 371
 372 void Yaz_PDU_Assoc::connect(IYaz_PDU_Observer *observer,
 373                             const char *addr)
 374 {
 375     logf (m_log, "Yaz_PDU_Assoc::connect %s", addr);
 376     close();
 377     m_PDU_Observer = observer;
 378     COMSTACK cs = comstack();
 379     void *ap = cs_straddr (cs, addr);
 380     if (!ap)
 381     {
 382         logf (m_log, "cs_straddr failed");
 383         return;
 384     }
 385     int res = cs_connect (cs, ap);
 386     logf (m_log, "Yaz_PDU_Assoc::connect fd=%d res=%d", cs_fileno(cs), res);
 387     m_socketObservable->addObserver(cs_fileno(cs), this);
 388     m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 389                                      YAZ_SOCKET_OBSERVE_EXCEPT|
 390                                      YAZ_SOCKET_OBSERVE_WRITE);
 391     if (res >= 0)
 392         m_state = Connecting;
 393     // if res < 0, then cs_connect failed immediately. state is Closed..
 394 }
 395
 396 void Yaz_PDU_Assoc::socket(IYaz_PDU_Observer *observer, int fd)
 397 {
 398     close();
 399     m_PDU_Observer = observer;
 400     if (fd >= 0)
 401     {
 402         CS_TYPE cs_type = tcpip_type;
 403         m_cs = cs_createbysocket(fd, cs_type, 0, PROTO_Z3950);
 404         m_state = Ready;
 405         m_socketObservable->addObserver(fd, this);
 406         m_socketObservable->maskObserver(this,
 407                                          YAZ_SOCKET_OBSERVE_READ|
 408                                          YAZ_SOCKET_OBSERVE_EXCEPT);
 409         m_socketObservable->timeoutObserver(this, m_idleTime);
 410     }
 411 }
 412
 413 #if 1
 414  // 1 = single-threaded
 415  // 0 = multi-threaded
 416
 417 // Single-threaded... Only useful for non-blocking handlers
 418 void Yaz_PDU_Assoc::childNotify(int fd)
 419 {
 420     // Clone PDU Observable (keep socket manager)
 421     IYaz_PDU_Observable *new_observable = clone();
 422
 423     // Clone PDU Observer
 424     IYaz_PDU_Observer *observer = m_PDU_Observer->clone(new_observable, fd);
 425
 426     // Attach new socket to it
 427     new_observable->socket(observer, fd);
 428 }
 429 #else
 430
 431 #include <yaz++/yaz-socket-manager.h>
 432
 433 #ifdef WIN32
 434 #include <process.h>
 435 #else
 436 #include <pthread.h>
 437 #endif
 438
 439 #ifdef WIN32
 440 void __cdecl
 441 #else
 442 void *
 443 #endif
 444     events(void *p)
 445 {
 446     Yaz_SocketManager *s = (Yaz_SocketManager *) p;
 447
 448     logf (LOG_LOG, "thread started");
 449     while (s->processEvent() > 0)
 450         ;
 451     logf (LOG_LOG, "thread finished");
 452 #ifdef WIN32
 453 #else
 454     return 0;
 455 #endif
 456 }
 457
 458 void Yaz_PDU_Assoc::childNotify(int fd)
 459 {
 460     Yaz_SocketManager *socket_observable = new Yaz_SocketManager;
 461     Yaz_PDU_Assoc *new_observable = new Yaz_PDU_Assoc (socket_observable);
 462
 463     /// Clone PDU Observer
 464     IYaz_PDU_Observer *observer = m_PDU_Observer->clone(new_observable, fd);
 465
 466     /// Attach new socket to it
 467     new_observable->socket(observer, fd);
 468
 469 #ifdef WIN32
 470     long t_id;
 471     t_id = _beginthread (events, 0, socket_observable);
 472     if (t_id == -1)
 473     {
 474         logf (LOG_FATAL|LOG_ERRNO, "_beginthread failed");
 475         exit (1);
 476     }
 477 #else
 478     pthread_t type;
 479
 480     int id = pthread_create (&type, 0, events, socket_observable);
 481     logf (LOG_LOG, "pthread_create returned id=%d", id);
 482 #endif
 483 }
 484 // Threads end
 485 #endif