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