ACE/netsvcs/lib/Token_Handler.cpp

   1 #include "ace/Get_Opt.h"
   2 #include "Token_Handler.h"
   3
   4 #if defined (ACE_HAS_TOKENS_LIBRARY)
   5
   6 #include "ace/Signal.h"
   7
   8 int
   9 ACE_Token_Acceptor::parse_args (int argc, ACE_TCHAR *argv[])
  10 {
  11   ACE_TRACE ("ACE_Token_Acceptor::parse_args");
  12
  13   u_short svc_port = ACE_DEFAULT_SERVER_PORT;
  14
  15   ACE_LOG_MSG->open (ACE_TEXT ("Token Service"));
  16
  17   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT ("p:"), 0);
  18
  19   for (int c; (c = get_opt ()) != -1; )
  20     {
  21       switch (c)
  22         {
  23         case 'p':
  24           svc_port = static_cast<u_short> (ACE_OS::atoi (get_opt.opt_arg ()));
  25           break;
  26         default:
  27           ACE_ERROR_RETURN ((LM_ERROR,
  28                              ACE_TEXT ("%n:\n[-p server-port]\n"), 1),
  29                             -1);
  30         }
  31     }
  32
  33   this->service_addr_.set (svc_port);
  34   return 0;
  35 }
  36
  37 int
  38 ACE_Token_Acceptor::init (int argc, ACE_TCHAR *argv[])
  39 {
  40   ACE_TRACE ("ACE_Token_Acceptor::init");
  41
  42   // Use the options hook to parse the command line arguments and set
  43   // options.
  44   this->parse_args (argc, argv);
  45
  46   // Set the acceptor endpoint into listen mode (use the Singleton
  47   // global Reactor...).
  48   if (this->open (this->service_addr_, ACE_Reactor::instance (),
  49                   0, 0, 0,
  50                   &this->scheduling_strategy_,
  51                   ACE_TEXT ("Token Server"),
  52                   ACE_TEXT ("ACE token service")) == -1)
  53     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%n: %p on port %d\n"),
  54                        ACE_TEXT ("acceptor::open failed"),
  55                        this->service_addr_.get_port_number ()), -1);
  56
  57   // Ignore SIGPIPE so that each <SVC_HANDLER> can handle this on its
  58   // own.
  59   ACE_Sig_Action sig ((ACE_SignalHandler) SIG_IGN, SIGPIPE);
  60   ACE_UNUSED_ARG (sig);
  61
  62   ACE_INET_Addr server_addr;
  63
  64   if (this->acceptor ().get_local_addr (server_addr) == -1)
  65     ACE_ERROR_RETURN
  66       ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("get_remote_addr")), -1);
  67
  68   ACE_DEBUG ((LM_DEBUG,
  69               ACE_TEXT ("starting up Token Server at port %d on handle %d\n"),
  70               server_addr.get_port_number (),
  71               this->acceptor ().get_handle ()));
  72   return 0;
  73 }
  74
  75 // The following is a "Factory" used by the ACE_Service_Config and
  76 // svc.conf file to dynamically initialize the state of the Naming
  77 // Server.
  78
  79 ACE_SVC_FACTORY_DEFINE (ACE_Token_Acceptor)
  80
  81 // Default constructor.
  82
  83 ACE_Token_Handler::ACE_Token_Handler (ACE_Thread_Manager *tm)
  84   : ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> (tm),
  85     collection_ (1),
  86     timeout_id_ (0)
  87 {
  88   ACE_TRACE ("ACE_Token_Handler::ACE_Token_Handler");
  89 }
  90
  91 // Create and send a reply to the client.
  92
  93 /* VIRTUAL */ int
  94 ACE_Token_Handler::send_reply (ACE_UINT32 err)
  95 {
  96   ACE_TRACE ("ACE_Token_Handler::send_reply");
  97   void *buf;
  98   size_t len;
  99   ssize_t n;
 100
 101   this->token_reply_.errnum (err);
 102
 103   len = this->token_reply_.encode (buf);
 104
 105   n = this->peer ().send (buf, len);
 106
 107   if (n != (ssize_t) len)
 108     ACE_ERROR_RETURN ((LM_ERROR,
 109                        ACE_TEXT ("%p, expected len = %d, actual len = %d\n"),
 110                        ACE_TEXT ("send failed"), len, n), -1);
 111   else
 112     return 0;
 113 }
 114
 115 // Acquire the token.
 116
 117 /* VIRTUAL */ int
 118 ACE_Token_Handler::acquire (ACE_Token_Proxy *proxy)
 119 {
 120   ACE_TRACE ("ACE_Token_Handler::acquire");
 121 #if 0
 122   ACE_DEBUG ((LM_DEBUG, "in acquire for client id = %s\n",
 123              proxy->client_id ()));
 124 #endif /* 0 */
 125
 126   // @@ add notify in token request reply
 127   if (proxy->acquire (0, 0, ACE_Synch_Options::asynch) == -1)
 128     {
 129       if (errno != EWOULDBLOCK)
 130         // bad bad bad
 131         return this->send_reply (errno);
 132
 133       // acquire would block
 134       if (request_options_[ACE_Synch_Options::USE_TIMEOUT] == 1)
 135         {
 136           // check for polling
 137           if ((request_options_.timeout ().sec () == 0) &&
 138               (request_options_.timeout ().usec () == 0))
 139             return this->send_reply (EWOULDBLOCK);
 140
 141           // schedule a timer
 142           this->timeout_id_ = this->reactor ()->schedule_timer
 143             (this, (void *) proxy, request_options_.timeout ());
 144           if (timeout_id_ == -1)
 145             {
 146               ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"),
 147                           ACE_TEXT ("schedule_timer")));
 148               return this->send_reply (errno);
 149             }
 150         }
 151       // send no reply.  wait until we acquire it or until the timer
 152       // goes off.
 153       return 0;
 154     }
 155   else // success
 156     return this->send_reply (0);
 157 }
 158
 159 // Try to acquire the token.  Never block.
 160
 161 /* VIRTUAL */ int
 162 ACE_Token_Handler::try_acquire (ACE_Token_Proxy *proxy)
 163 {
 164   ACE_TRACE ("ACE_Token_Handler::try_acquire");
 165
 166 #if 0
 167   ACE_DEBUG ((LM_DEBUG, "in try_acquire for client id = %s\n",
 168              proxy->client_id ()));
 169 #endif /* 0 */
 170
 171   // @@ add notify in token request reply
 172   if (proxy->tryacquire () == -1)
 173     return this->send_reply (errno);
 174   else
 175     return this->send_reply (0);
 176 }
 177
 178 // Release the token and allow the next client that is waiting to
 179 // proceed.
 180
 181 /* VIRTUAL */ int
 182 ACE_Token_Handler::release (ACE_Token_Proxy *proxy)
 183 {
 184   ACE_TRACE ("ACE_Token_Handler::release");
 185 #if 0
 186   ACE_DEBUG ((LM_DEBUG,
 187               "in release for client id = %s\n",
 188               proxy->client_id ()));
 189 #endif /* 0 */
 190
 191   if (proxy->release (ACE_Synch_Options::asynch) == -1)
 192     // oops, it failed
 193     return this->send_reply (ACE_LOG_MSG->errnum ());
 194
 195   // success
 196   if (this->timeout_id_ != 0)
 197     {
 198       this->reactor ()->cancel_timer (timeout_id_);
 199       this->timeout_id_ = 0;
 200     }
 201
 202   return this->send_reply (0);
 203 }
 204
 205 // Yield the token if any clients are waiting, otherwise keep the
 206 // token.
 207
 208 /* VIRTUAL */ int
 209 ACE_Token_Handler::renew (ACE_Token_Proxy *proxy)
 210 {
 211   ACE_TRACE ("ACE_Token_Handler::renew");
 212
 213 #if 0
 214   ACE_DEBUG ((LM_DEBUG, "in renew for client id = %s\n",
 215              proxy->client_id ()));
 216 #endif /* 0 */
 217
 218   if (proxy->renew (token_request_.requeue_position (),
 219                     ACE_Synch_Options::asynch) == -1)
 220     {
 221       int result = ACE_LOG_MSG->errnum ();
 222       if (result != EWOULDBLOCK)
 223         // bad bad bad
 224         return this->send_reply (result);
 225
 226       // acquire would block
 227       if (request_options_[ACE_Synch_Options::USE_TIMEOUT] == 1)
 228         {
 229           this->timeout_id_ = this->reactor ()->schedule_timer
 230             (this, 0, request_options_.timeout ());
 231           if (timeout_id_ == -1)
 232             {
 233               ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"),
 234                           ACE_TEXT ("schedule_timer")));
 235               return this->send_reply (ACE_LOG_MSG->errnum ());
 236             }
 237         }
 238       // Send no reply.  wait until we acquire it or until the timer
 239       // goes off.
 240       return 0;
 241     }
 242   else
 243     // Success, we still hold the token.
 244     return this->send_reply (0);
 245 }
 246
 247 /* VIRTUAL */ int
 248 ACE_Token_Handler::remove (ACE_Token_Proxy * /* proxy */)
 249 {
 250   ACE_TRACE ("ACE_Token_Handler::remove");
 251 #if 0
 252   ACE_DEBUG ((LM_DEBUG, "in remove for client id = %s\n",
 253              proxy->client_id ()));
 254 #endif /* 0 */
 255   ACE_ERROR
 256     ((LM_ERROR,
 257       ACE_TEXT ("sorry: ACE_Token_Handler::remove() is not implemented")));
 258
 259   return this->send_reply (ENOTSUP);
 260 }
 261
 262 // Enable clients to limit the amount of time they'll wait for a
 263 // token.
 264
 265 /* VIRTUAL */ int
 266 ACE_Token_Handler::handle_timeout (const ACE_Time_Value &,
 267                                    const void *tp)
 268 {
 269   ACE_TRACE ("ACE_Token_Handler::handle_timeout");
 270
 271   this->timeout_id_ = 0;
 272
 273   // @@ add a try acquire here!
 274   // Try to acquire the token, but if we can't get it immediately
 275   // then abandon the wait.
 276   //  if (this->try_acquire (&token_entry) == -1)
 277   //    return this->abandon (token_entry);
 278
 279   ACE_Token_Proxy *proxy = (ACE_Token_Proxy *) tp;
 280
 281 #if 0
 282   ACE_DEBUG ((LM_DEBUG, "in handle_timeout for client id = %s\n",
 283              proxy->client_id ()));
 284 #endif /* 0 */
 285
 286   // Remove ourselves from the waiter list.
 287   proxy->release ();
 288
 289   this->send_reply (ETIME);
 290   return 0;
 291 }
 292
 293 // Dispatch the appropriate operation to handle the client request.
 294
 295 ACE_Token_Proxy *
 296 ACE_Token_Handler::get_proxy (void)
 297 {
 298   ACE_TRACE ("ACE_Token_Handler::get_proxy");
 299
 300   // See if the proxy already exists in the collection.
 301   ACE_Token_Proxy *proxy = collection_.is_member (token_request_.token_name ());
 302
 303   // If not, create one.
 304   if (proxy == 0)
 305     {
 306       proxy = this->create_proxy ();
 307
 308       // Put the new_proxy in this client_id's collection.
 309       if (collection_.insert (*proxy) == -1)
 310         ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("insert failed\n")), 0);
 311
 312       // Delete our copy (one was created in the collection).
 313       delete proxy;
 314       proxy = collection_.is_member (token_request_.token_name ());
 315
 316       if (proxy == 0)
 317         ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("is_member failed\n")), 0);
 318
 319       // Set the client_id (it was set to 1 since we're
 320       // single-threaded.
 321       proxy->client_id (token_request_.client_id ());
 322     }
 323
 324   return proxy;
 325 }
 326
 327 ACE_Token_Proxy *
 328 ACE_Token_Handler::create_proxy (void)
 329 {
 330   ACE_TRACE ("ACE_Token_Handler::create_proxy");
 331
 332   ACE_Token_Proxy *proxy;
 333
 334   switch (token_request_.token_type ())
 335     {
 336     case ACE_Tokens::RWLOCK:
 337       if (token_request_.proxy_type () == ACE_RW_Token::READER)
 338         ACE_NEW_RETURN (proxy,
 339                         ACE_TS_RLock (token_request_.token_name (), this),
 340                         0);
 341       else
 342         ACE_NEW_RETURN (proxy,
 343                         ACE_TS_WLock (token_request_.token_name (), this),
 344                         0);
 345       break;
 346     case ACE_Tokens::MUTEX:
 347       ACE_NEW_RETURN (proxy,
 348                       ACE_TS_Mutex (token_request_.token_name (), this),
 349                       0);
 350       break;
 351     default:
 352       // Nonexistent token type.
 353       errno = EINVAL;
 354       return 0;
 355     }
 356
 357   // Check for failed new.
 358   if (proxy == 0)
 359     errno = ENOMEM;
 360
 361   return proxy;
 362 }
 363
 364 int
 365 ACE_Token_Handler::dispatch (void)
 366 {
 367   ACE_TRACE ("ACE_Token_Handler::dispatch");
 368   ACE_Token_Proxy *proxy = this->get_proxy ();
 369
 370   if (proxy == 0)
 371     return -1;
 372
 373   // Dispatch the appropriate request.
 374   switch (this->token_request_.operation_type ())
 375     {
 376     case ACE_Token_Request::ACQUIRE:
 377       return this->acquire (proxy);
 378     case ACE_Token_Request::TRY_ACQUIRE:
 379       return this->try_acquire (proxy);
 380     case ACE_Token_Request::RELEASE:
 381       return this->release (proxy);
 382     case ACE_Token_Request::RENEW:
 383       return this->renew (proxy);
 384     case ACE_Token_Request::REMOVE:
 385       return this->remove (proxy);
 386     default:
 387       ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("invalid type = %d\n"),
 388                         this->token_request_.operation_type ()), -1);
 389       /* NOTREACHED */
 390     }
 391 }
 392
 393 // Receive, frame, and decode the client's request.
 394 // Note, this method should use non-blocking I/O.
 395
 396 /* VIRTUAL */ int
 397 ACE_Token_Handler::recv_request (void)
 398 {
 399   ACE_TRACE ("ACE_Token_Handler::recv_request");
 400   ssize_t n;
 401
 402   // Read the first 4 bytes to get the length of the message
 403   // This implementation assumes that the first 4 bytes are
 404   // the length of the message.
 405   n = this->peer ().recv ((void *) &this->token_request_,
 406                           sizeof (ACE_UINT32));
 407
 408   switch (n)
 409     {
 410     case -1:
 411       /* FALLTHROUGH */
 412     default:
 413       ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p got %d bytes, expected %d bytes\n"),
 414                   ACE_TEXT ("recv failed"), n, sizeof (ACE_UINT32)));
 415       /* FALLTHROUGH */
 416     case 0:
 417       // We've shutdown unexpectedly, let's abandon the connection.
 418       this->abandon (0);
 419       return -1;
 420       /* NOTREACHED */
 421     case sizeof (ACE_UINT32):
 422       {
 423         // Transform the length into host byte order.
 424         ssize_t length = this->token_request_.length ();
 425
 426         // Do a sanity check on the length of the message.
 427         if (length > (ssize_t) sizeof this->token_request_)
 428           {
 429             ACE_ERROR ((LM_ERROR, ACE_TEXT ("length %d too long\n"), length));
 430             return this->abandon (1);
 431           }
 432
 433         // Receive the rest of the request message.
 434         // @@ beware of blocking read!!!.
 435         n = this->peer ().recv ((void *) (((char *) &this->token_request_)
 436                                           + sizeof (ACE_UINT32)),
 437                                 length - sizeof (ACE_UINT32));
 438
 439         // Subtract off the size of the part we skipped over...
 440         if (n != (length - (ssize_t) sizeof (ACE_UINT32)))
 441           {
 442             ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p expected %d, got %d\n"),
 443                         ACE_TEXT ("invalid length"), length, n));
 444             return this->abandon (1);
 445           }
 446
 447         // Decode the request into host byte order.
 448         if (this->token_request_.decode () == -1)
 449           {
 450             ACE_ERROR
 451               ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("decode failed")));
 452             return this->abandon (1);
 453           }
 454
 455         // if (OS::debug)
 456         this->token_request_.dump ();
 457       }
 458     }
 459   return 0;
 460 }
 461
 462 // Callback method invoked by the ACE_Reactor when
 463 // events arrive from the client.
 464
 465 /* VIRTUAL */ int
 466 ACE_Token_Handler::handle_input (ACE_HANDLE)
 467 {
 468   ACE_TRACE ("ACE_Token_Handler::handle_input");
 469
 470 #if 0
 471   ACE_DEBUG ((LM_DEBUG, "****************** in handle_input\n"));
 472 #endif /* 0 */
 473
 474   if (this->recv_request () == -1)
 475     return -1;
 476   else
 477     return this->dispatch ();
 478 }
 479
 480 void
 481 ACE_Token_Handler::sleep_hook (void)
 482 {
 483   ACE_TRACE ("ACE_Token_Handler::sleep_hook");
 484   // @@ what should we do?
 485   return;
 486 }
 487
 488 void
 489 ACE_Token_Handler::token_acquired (ACE_TPQ_Entry *)
 490 {
 491   ACE_TRACE ("ACE_Token_Handler::token_acquired");
 492
 493   if (this->timeout_id_ != 0)
 494     {
 495       this->reactor ()->cancel_timer (this->timeout_id_);
 496       this->timeout_id_ = 0;
 497     }
 498
 499   this->send_reply (0);
 500 }
 501
 502 int
 503 ACE_Token_Handler::abandon (int send_error)
 504 {
 505   ACE_TRACE ("ACE_Token_Handler::abandon");
 506
 507   // Release ownership or remove us from the waiter list.
 508   if (this->timeout_id_ != 0)
 509     {
 510       this->reactor ()->cancel_timer (timeout_id_);
 511       this->timeout_id_ = 0;
 512     }
 513
 514   // @@ release all tokens
 515   collection_.release ();
 516
 517   if (send_error)
 518     return this->send_reply (EIO);
 519   else
 520     return -1;
 521 }
 522
 523 // ************************************************************
 524 // ************************************************************
 525 // ************************************************************
 526
 527 ACE_TS_Mutex::ACE_TS_Mutex (const ACE_TCHAR *name,
 528                             ACE_Token_Handler *th)
 529 : ACE_Local_Mutex (name, 0, 1), // The 1 is debug.
 530   th_ (th)
 531 {
 532   ACE_TRACE ("ACE_TS_Mutex::ACE_TS_Mutex");
 533 }
 534
 535 ACE_TS_Mutex::ACE_TS_Mutex (const ACE_TS_Mutex &m)
 536 : ACE_Local_Mutex (m),
 537   th_ (m.th_)
 538 {
 539   ACE_TRACE ("ACE_TS_Mutex::ACE_TS_Mutex");
 540   this->open (m.name (), m.ignore_deadlock_, m.debug_);
 541 }
 542
 543 void
 544 ACE_TS_Mutex::sleep_hook (void)
 545 {
 546   ACE_TRACE ("ACE_TS_Mutex::sleep_hook");
 547   th_->sleep_hook ();
 548   return;
 549 }
 550
 551 void
 552 ACE_TS_Mutex::token_acquired (ACE_TPQ_Entry *e)
 553 {
 554   ACE_TRACE ("ACE_TS_Mutex::token_acquired");
 555   // Notify the token handler.
 556   th_->token_acquired (e);
 557   return;
 558 }
 559
 560 ACE_Token_Proxy *
 561 ACE_TS_Mutex::clone (void) const
 562 {
 563   ACE_TRACE ("ACE_TS_Mutex::clone");
 564   ACE_Token_Proxy *temp;
 565   ACE_NEW_RETURN (temp, ACE_TS_Mutex (*this), 0);
 566   return temp;
 567 }
 568
 569 // ************************************************************
 570
 571 ACE_TS_RLock::ACE_TS_RLock (const ACE_TCHAR *name,
 572                             ACE_Token_Handler *th)
 573 : ACE_Local_RLock (name, 0, 1), // The 1 is debug.
 574   th_ (th)
 575 {
 576   ACE_TRACE ("ACE_TS_RLock::ACE_TS_RLock");
 577 }
 578
 579 ACE_TS_RLock::ACE_TS_RLock (const ACE_TS_RLock &r)
 580 : ACE_Local_RLock (r),
 581   th_ (r.th_)
 582 {
 583   ACE_TRACE ("ACE_TS_RLock::ACE_TS_RLock");
 584   this->open (r.name (), r.ignore_deadlock_, r.debug_);
 585 }
 586
 587 void
 588 ACE_TS_RLock::sleep_hook (void)
 589 {
 590   ACE_TRACE ("ACE_TS_RLock::sleep_hook");
 591   th_->sleep_hook ();
 592   return;
 593 }
 594
 595 void
 596 ACE_TS_RLock::token_acquired (ACE_TPQ_Entry *e)
 597 {
 598   ACE_TRACE ("ACE_TS_RLock::token_acquired");
 599   // Notify the token handler.
 600   th_->token_acquired (e);
 601   return;
 602 }
 603
 604 ACE_Token_Proxy *
 605 ACE_TS_RLock::clone (void) const
 606 {
 607   ACE_TRACE ("ACE_TS_RLock::clone");
 608   ACE_Token_Proxy *temp;
 609
 610   ACE_NEW_RETURN (temp, ACE_TS_RLock (*this), 0);
 611   return temp;
 612 }
 613
 614 // ************************************************************
 615
 616 ACE_TS_WLock::ACE_TS_WLock (const ACE_TCHAR *name,
 617                             ACE_Token_Handler *th)
 618 : ACE_Local_WLock (name, 0, 1), // The 1 is debug.
 619   th_ (th)
 620 {
 621   ACE_TRACE ("ACE_TS_WLock::ACE_TS_WLock");
 622 }
 623
 624 ACE_TS_WLock::ACE_TS_WLock (const ACE_TS_WLock &w)
 625 : ACE_Local_WLock (w),
 626   th_ (w.th_)
 627 {
 628   ACE_TRACE ("ACE_TS_WLock::ACE_TS_WLock");
 629   this->open (w.name (), w.ignore_deadlock_, w.debug_);
 630 }
 631
 632 void
 633 ACE_TS_WLock::sleep_hook (void)
 634 {
 635   ACE_TRACE ("ACE_TS_WLock::sleep_hook");
 636   th_->sleep_hook ();
 637   return;
 638 }
 639
 640 void
 641 ACE_TS_WLock::token_acquired (ACE_TPQ_Entry *e)
 642 {
 643   ACE_TRACE ("ACE_TS_WLock::token_acquired");
 644   // Notify the token handler.
 645   th_->token_acquired (e);
 646   return;
 647 }
 648
 649 ACE_Token_Proxy *
 650 ACE_TS_WLock::clone (void) const
 651 {
 652   ACE_TRACE ("ACE_TS_WLock::clone");
 653   ACE_Token_Proxy *temp;
 654
 655   ACE_NEW_RETURN (temp, ACE_TS_WLock (*this), 0);
 656   return temp;
 657 }
 658
 659 #endif /* ACE_HAS_TOKENS_LIBRARY */