src/yaz-pdu-assoc.cpp

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