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