src/yaz-pdu-assoc.cpp

   1 /*
   2  * Copyright (c) 1998-2004, Index Data.
   3  * See the file LICENSE for details.
   4  *
   5  * $Id: yaz-pdu-assoc.cpp,v 1.47 2008-01-21 15:57:27 adam Exp $
   6  */
   7
   8 #include <assert.h>
   9 #include <string.h>
  10 #include <yaz/log.h>
  11 #include <yaz/tcpip.h>
  12
  13 #include <yazpp/pdu-assoc.h>
  14
  15 using namespace yazpp_1;
  16
  17 void PDU_Assoc::init(ISocketObservable *socketObservable)
  18 {
  19     m_state = Closed;
  20     m_cs = 0;
  21     m_socketObservable = socketObservable;
  22     m_PDU_Observer = 0;
  23     m_queue_out = 0;
  24     m_queue_in = 0;
  25     m_input_buf = 0;
  26     m_input_len = 0;
  27     m_children = 0;
  28     m_parent = 0;
  29     m_next = 0;
  30     m_destroyed = 0;
  31     m_idleTime = 0;
  32     m_log = YLOG_DEBUG;
  33     m_session_is_dead = false;
  34 }
  35
  36 PDU_Assoc::PDU_Assoc(ISocketObservable *socketObservable)
  37 {
  38     init (socketObservable);
  39 }
  40
  41 PDU_Assoc::PDU_Assoc(ISocketObservable *socketObservable,
  42                      COMSTACK cs)
  43 {
  44     init(socketObservable);
  45     m_cs = cs;
  46     unsigned mask = 0;
  47     if (cs->io_pending & CS_WANT_WRITE)
  48         mask |= SOCKET_OBSERVE_WRITE;
  49     if (cs->io_pending & CS_WANT_READ)
  50         mask |= SOCKET_OBSERVE_READ;
  51     m_socketObservable->addObserver(cs_fileno(cs), this);
  52     if (!mask)
  53     {
  54         yaz_log (m_log, "new PDU_Assoc. Ready");
  55         m_state = Ready;
  56         flush_PDU();
  57     }
  58     else
  59     {
  60         yaz_log (m_log, "new PDU_Assoc. Accepting");
  61         // assume comstack is accepting...
  62         m_state = Accepting;
  63         m_socketObservable->addObserver(cs_fileno(cs), this);
  64         yaz_log(m_log, "maskObserver 1");
  65         m_socketObservable->maskObserver(this,
  66                                          mask |SOCKET_OBSERVE_EXCEPT);
  67     }
  68 }
  69
  70
  71 IPDU_Observable *PDU_Assoc::clone()
  72 {
  73     PDU_Assoc *copy = new PDU_Assoc(m_socketObservable);
  74     return copy;
  75 }
  76
  77 void PDU_Assoc::socketNotify(int event)
  78 {
  79     yaz_log (m_log, "PDU_Assoc::socketNotify p=%p state=%d event = %d",
  80           this, m_state, event);
  81     if (event & SOCKET_OBSERVE_EXCEPT)
  82     {
  83         shutdown();
  84         m_PDU_Observer->failNotify();
  85         return;
  86     }
  87     else if (event & SOCKET_OBSERVE_TIMEOUT)
  88     {
  89         m_PDU_Observer->timeoutNotify();
  90         return;
  91     }
  92     switch (m_state)
  93     {
  94     case Accepting:
  95         if (!cs_accept (m_cs))
  96         {
  97             yaz_log (m_log, "PDU_Assoc::cs_accept failed");
  98             m_cs = 0;
  99             shutdown();
 100             m_PDU_Observer->failNotify();
 101         }
 102         else
 103         {
 104             unsigned mask = 0;
 105             if (m_cs->io_pending & CS_WANT_WRITE)
 106                 mask |= SOCKET_OBSERVE_WRITE;
 107             if (m_cs->io_pending & CS_WANT_READ)
 108                 mask |= SOCKET_OBSERVE_READ;
 109             if (!mask)
 110             {   // accept is complete. turn to ready state and write if needed
 111                 m_state = Ready;
 112                 flush_PDU();
 113             }
 114             else
 115             {   // accept still incomplete.
 116                 yaz_log(m_log, "maskObserver 2");
 117                 m_socketObservable->maskObserver(this,
 118                                              mask|SOCKET_OBSERVE_EXCEPT);
 119             }
 120         }
 121         break;
 122     case Connecting:
 123         if (event & SOCKET_OBSERVE_READ &&
 124             event & SOCKET_OBSERVE_WRITE)
 125         {
 126             // For Unix: if both read and write is set, then connect failed.
 127             shutdown();
 128             m_PDU_Observer->failNotify();
 129         }
 130         else
 131         {
 132             yaz_log (m_log, "cs_rcvconnect");
 133             int res = cs_rcvconnect (m_cs);
 134             if (res == 1)
 135             {
 136                 unsigned mask = SOCKET_OBSERVE_EXCEPT;
 137                 if (m_cs->io_pending & CS_WANT_WRITE)
 138                     mask |= SOCKET_OBSERVE_WRITE;
 139                 if (m_cs->io_pending & CS_WANT_READ)
 140                     mask |= SOCKET_OBSERVE_READ;
 141                 yaz_log(m_log, "maskObserver 3");
 142                 m_socketObservable->maskObserver(this, mask);
 143             }
 144             else
 145             {
 146                 m_state = Ready;
 147                 if (m_PDU_Observer)
 148                     m_PDU_Observer->connectNotify();
 149                 flush_PDU();
 150             }
 151         }
 152         break;
 153     case Listen:
 154         if (event & SOCKET_OBSERVE_READ)
 155         {
 156             int res;
 157             COMSTACK new_line;
 158
 159             if ((res = cs_listen(m_cs, 0, 0)) == 1)
 160                 return;
 161             if (res < 0)
 162             {
 163                 yaz_log(YLOG_FATAL|YLOG_ERRNO, "cs_listen failed");
 164                 return;
 165             }
 166             if (!(new_line = cs_accept(m_cs)))
 167                 return;
 168             /* 1. create socket-manager
 169                2. create pdu-assoc
 170                3. create top-level object
 171                     setup observer for child fileid in pdu-assoc
 172                4. start thread
 173             */
 174             yaz_log (m_log, "new session: parent fd=%d child fd=%d",
 175                      cs_fileno(m_cs), cs_fileno(new_line));
 176             childNotify (new_line);
 177         }
 178         break;
 179     case Writing:
 180         if (event & (SOCKET_OBSERVE_READ|SOCKET_OBSERVE_WRITE))
 181             flush_PDU();
 182         break;
 183     case Ready:
 184         if (event & (SOCKET_OBSERVE_READ|SOCKET_OBSERVE_WRITE))
 185         {
 186             do
 187             {
 188                 int res = cs_get (m_cs, &m_input_buf, &m_input_len);
 189                 if (res == 1)
 190                 {
 191                     unsigned mask = SOCKET_OBSERVE_EXCEPT;
 192                     if (m_cs->io_pending & CS_WANT_WRITE)
 193                         mask |= SOCKET_OBSERVE_WRITE;
 194                     if (m_cs->io_pending & CS_WANT_READ)
 195                         mask |= SOCKET_OBSERVE_READ;
 196                     yaz_log(m_log, "maskObserver 4");
 197                     m_socketObservable->maskObserver(this, mask);
 198                     return;
 199                 }
 200                 else if (res <= 0)
 201                 {
 202                     yaz_log (m_log, "PDU_Assoc::Connection closed by peer");
 203                     shutdown();
 204                     if (m_PDU_Observer)
 205                         m_PDU_Observer->failNotify(); // problem here..
 206                     return;
 207                 }
 208                 // lock it, so we know if recv_PDU deletes it.
 209                 int destroyed = 0;
 210                 m_destroyed = &destroyed;
 211
 212                 if (!m_PDU_Observer)
 213                     return;
 214 #if 0
 215                 PDU_Queue **pq = &m_queue_in;
 216                 while (*pq)
 217                     pq = &(*pq)->m_next;
 218
 219                 *pq = new PDU_Queue(m_input_buf, res);
 220 #else
 221                 m_PDU_Observer->recv_PDU(m_input_buf, res);
 222 #endif
 223                 if (destroyed)   // it really was destroyed, return now.
 224                     return;
 225                 m_destroyed = 0;
 226             } while (m_cs && cs_more (m_cs));
 227             if (m_cs && m_state == Ready)
 228             {
 229                 yaz_log(m_log, "maskObserver 5");
 230                 m_socketObservable->maskObserver(this,
 231                                                  SOCKET_OBSERVE_EXCEPT|
 232                                                  SOCKET_OBSERVE_READ);
 233             }
 234         }
 235         break;
 236     case Closed:
 237         yaz_log (m_log, "CLOSING state=%d event was %d", m_state, event);
 238         shutdown();
 239         m_PDU_Observer->failNotify();
 240         break;
 241     default:
 242         yaz_log (m_log, "Unknown state=%d event was %d", m_state, event);
 243         shutdown();
 244         m_PDU_Observer->failNotify();
 245     }
 246 }
 247
 248 void PDU_Assoc::close_session()
 249 {
 250     m_session_is_dead = true;
 251     if (!m_queue_out)
 252     {
 253         shutdown();
 254         m_PDU_Observer->failNotify();
 255     }
 256 }
 257
 258 void PDU_Assoc::shutdown()
 259 {
 260     PDU_Assoc *ch;
 261     for (ch = m_children; ch; ch = ch->m_next)
 262         ch->shutdown();
 263
 264     m_socketObservable->deleteObserver(this);
 265     m_state = Closed;
 266     if (m_cs)
 267     {
 268         yaz_log (m_log, "PDU_Assoc::close fd=%d", cs_fileno(m_cs));
 269         cs_close (m_cs);
 270     }
 271     m_cs = 0;
 272     while (m_queue_out)
 273     {
 274         PDU_Queue *q_this = m_queue_out;
 275         m_queue_out = m_queue_out->m_next;
 276         delete q_this;
 277     }
 278     xfree (m_input_buf);
 279     m_input_buf = 0;
 280     m_input_len = 0;
 281 }
 282
 283 void PDU_Assoc::destroy()
 284 {
 285     shutdown();
 286
 287     if (m_destroyed)
 288         *m_destroyed = 1;
 289     PDU_Assoc **c;
 290
 291     // delete from parent's child list (if any)
 292     if (m_parent)
 293     {
 294         c = &m_parent->m_children;
 295         while (*c != this)
 296         {
 297             assert (*c);
 298             c = &(*c)->m_next;
 299         }
 300         *c = (*c)->m_next;
 301     }
 302     // delete all children ...
 303     c = &m_children;
 304     while (*c)
 305     {
 306         PDU_Assoc *here = *c;
 307         *c = (*c)->m_next;
 308         here->m_parent = 0;
 309         delete here;
 310     }
 311     yaz_log (m_log, "PDU_Assoc::destroy this=%p", this);
 312 }
 313
 314 PDU_Assoc::PDU_Queue::PDU_Queue(const char *buf, int len)
 315 {
 316     m_buf = (char *) xmalloc (len);
 317     memcpy (m_buf, buf, len);
 318     m_len = len;
 319     m_next = 0;
 320 }
 321
 322 PDU_Assoc::PDU_Queue::~PDU_Queue()
 323 {
 324     xfree (m_buf);
 325 }
 326
 327 int PDU_Assoc::flush_PDU()
 328 {
 329     int r;
 330
 331     if (m_state != Ready && m_state != Writing)
 332     {
 333         yaz_log (m_log, "YAZ_PDU_Assoc::flush_PDU, not ready");
 334         return 1;
 335     }
 336     PDU_Queue *q = m_queue_out;
 337     if (!q)
 338     {
 339         m_state = Ready;
 340         yaz_log (m_log, "YAZ_PDU_Assoc::flush_PDU queue empty");
 341         yaz_log(m_log, "maskObserver 6");
 342         m_socketObservable->maskObserver(this, SOCKET_OBSERVE_READ|
 343                                          SOCKET_OBSERVE_WRITE|
 344                                          SOCKET_OBSERVE_EXCEPT);
 345         if (m_session_is_dead)
 346         {
 347             shutdown();
 348             m_PDU_Observer->failNotify();
 349         }
 350         return 0;
 351     }
 352     r = cs_put (m_cs, q->m_buf, q->m_len);
 353     if (r < 0)
 354     {
 355         yaz_log (m_log, "PDU_Assoc::flush_PDU cs_put failed");
 356         shutdown();
 357         m_PDU_Observer->failNotify();
 358         return r;
 359     }
 360     if (r == 1)
 361     {
 362         unsigned mask = SOCKET_OBSERVE_EXCEPT;
 363         m_state = Writing;
 364         if (m_cs->io_pending & CS_WANT_WRITE)
 365             mask |= SOCKET_OBSERVE_WRITE;
 366         if (m_cs->io_pending & CS_WANT_READ)
 367             mask |= SOCKET_OBSERVE_READ;
 368
 369         mask |= SOCKET_OBSERVE_WRITE;
 370         yaz_log(m_log, "maskObserver 7");
 371         m_socketObservable->maskObserver(this, mask);
 372         yaz_log (m_log, "PDU_Assoc::flush_PDU cs_put %d bytes fd=%d (inc)",
 373                  q->m_len, cs_fileno(m_cs));
 374         return r;
 375     }
 376     yaz_log (m_log, "PDU_Assoc::flush_PDU cs_put %d bytes", q->m_len);
 377     // whole packet sent... delete this and proceed to next ...
 378     m_queue_out = q->m_next;
 379     delete q;
 380     // don't select on write if queue is empty ...
 381     if (!m_queue_out)
 382     {
 383         m_state = Ready;
 384         yaz_log(m_log, "maskObserver 8");
 385         m_socketObservable->maskObserver(this, SOCKET_OBSERVE_READ|
 386                                          SOCKET_OBSERVE_EXCEPT);
 387     }
 388     return r;
 389 }
 390
 391 int PDU_Assoc::send_PDU(const char *buf, int len)
 392 {
 393     yaz_log (m_log, "PDU_Assoc::send_PDU");
 394     PDU_Queue **pq = &m_queue_out;
 395     int is_idle = (*pq ? 0 : 1);
 396
 397     if (!m_cs)
 398     {
 399         yaz_log (m_log, "PDU_Assoc::send_PDU failed, m_cs == 0");
 400         return -1;
 401     }
 402     while (*pq)
 403         pq = &(*pq)->m_next;
 404     *pq = new PDU_Queue(buf, len);
 405     if (is_idle)
 406         return flush_PDU ();
 407     else
 408         yaz_log (m_log, "PDU_Assoc::cannot send_PDU fd=%d",
 409                  cs_fileno(m_cs));
 410     return 0;
 411 }
 412
 413 COMSTACK PDU_Assoc::comstack(const char *type_and_host, void **vp)
 414 {
 415     return cs_create_host(type_and_host, 2, vp);
 416 }
 417
 418 int PDU_Assoc::listen(IPDU_Observer *observer, const char *addr)
 419 {
 420     shutdown();
 421
 422     m_PDU_Observer = observer;
 423     void *ap;
 424     m_cs = comstack(addr, &ap);
 425
 426     if (!m_cs)
 427         return -1;
 428     if (cs_bind(m_cs, ap, CS_SERVER) < 0)
 429         return -2;
 430     m_socketObservable->addObserver(cs_fileno(m_cs), this);
 431     yaz_log(m_log, "maskObserver 9");
 432     m_socketObservable->maskObserver(this, SOCKET_OBSERVE_READ|
 433                                      SOCKET_OBSERVE_EXCEPT);
 434     yaz_log (m_log, "PDU_Assoc::listen ok fd=%d", cs_fileno(m_cs));
 435     m_state = Listen;
 436     return 0;
 437 }
 438
 439 void PDU_Assoc::idleTime(int idleTime)
 440 {
 441     m_idleTime = idleTime;
 442     yaz_log (m_log, "PDU_Assoc::idleTime(%d)", idleTime);
 443     m_socketObservable->timeoutObserver(this, m_idleTime);
 444 }
 445
 446 int PDU_Assoc::connect(IPDU_Observer *observer, const char *addr)
 447 {
 448     yaz_log (m_log, "PDU_Assoc::connect %s", addr);
 449     shutdown();
 450     m_PDU_Observer = observer;
 451     void *ap;
 452     m_cs = comstack(addr, &ap);
 453     if (!m_cs)
 454         return -1;
 455     int res = cs_connect (m_cs, ap);
 456     yaz_log (m_log, "PDU_Assoc::connect fd=%d res=%d", cs_fileno(m_cs),
 457              res);
 458     m_socketObservable->addObserver(cs_fileno(m_cs), this);
 459
 460     if (res == 0)
 461     {   // Connect complete
 462         m_state = Connecting;
 463         unsigned mask = SOCKET_OBSERVE_EXCEPT;
 464         mask |= SOCKET_OBSERVE_WRITE;
 465         mask |= SOCKET_OBSERVE_READ;
 466         yaz_log(m_log, "maskObserver 11");
 467         m_socketObservable->maskObserver(this, mask);
 468     }
 469     else if (res > 0)
 470     {   // Connect pending
 471         m_state = Connecting;
 472         unsigned mask = SOCKET_OBSERVE_EXCEPT;
 473         if (m_cs->io_pending & CS_WANT_WRITE)
 474             mask |= SOCKET_OBSERVE_WRITE;
 475         if (m_cs->io_pending & CS_WANT_READ)
 476             mask |= SOCKET_OBSERVE_READ;
 477         yaz_log(m_log, "maskObserver 11");
 478         m_socketObservable->maskObserver(this, mask);
 479     }
 480     else
 481     {   // Connect failed immediately
 482         // Since m_state is Closed we can distinguish this case from
 483         // normal connect in socketNotify handler
 484         yaz_log(m_log, "maskObserver 12");
 485         m_socketObservable->maskObserver(this, SOCKET_OBSERVE_WRITE|
 486                                          SOCKET_OBSERVE_EXCEPT);
 487     }
 488     return 0;
 489 }
 490
 491 // Single-threaded... Only useful for non-blocking handlers
 492 void PDU_Assoc::childNotify(COMSTACK cs)
 493 {
 494     PDU_Assoc *new_observable =
 495         new PDU_Assoc (m_socketObservable, cs);
 496
 497     // Clone PDU Observer
 498     new_observable->m_PDU_Observer = m_PDU_Observer->sessionNotify
 499         (new_observable, cs_fileno(cs));
 500
 501     if (!new_observable->m_PDU_Observer)
 502     {
 503         new_observable->shutdown();
 504         delete new_observable;
 505         return;
 506     }
 507     new_observable->m_next = m_children;
 508     m_children = new_observable;
 509     new_observable->m_parent = this;
 510 }
 511
 512 const char*PDU_Assoc::getpeername()
 513 {
 514     if (!m_cs)
 515         return 0;
 516     return cs_addrstr(m_cs);
 517 }
 518 /*
 519  * Local variables:
 520  * c-basic-offset: 4
 521  * indent-tabs-mode: nil
 522  * End:
 523  * vim: shiftwidth=4 tabstop=8 expandtab
 524  */
 525