src/apps/cortex/NodeManager/NodeGroup.cpp

   1 /*
   2  * Copyright (c) 1999-2000, Eric Moon.
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  *
   9  * 1. Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions, and the following disclaimer.
  11  *
  12  * 2. Redistributions in binary form must reproduce the above copyright
  13  *    notice, this list of conditions, and the following disclaimer in the
  14  *    documentation and/or other materials provided with the distribution.
  15  *
  16  * 3. The name of the author may not be used to endorse or promote products
  17  *    derived from this software without specific prior written permission.
  18  *
  19  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR "AS IS" AND ANY EXPRESS OR
  20  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  21  * OF TITLE, NON-INFRINGEMENT, MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  22  * PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
  23  * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  24  * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  25  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  26  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
  27  * TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  28  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29  */
  30
  31
  32 // NodeGroup.cpp
  33
  34 #include "NodeGroup.h"
  35 //#include "NodeGroup_transport_thread.h"
  36
  37 #include "NodeManager.h"
  38 #include "NodeRef.h"
  39
  40 #include <MediaRoster.h>
  41 #include <OS.h>
  42 #include <TimeSource.h>
  43
  44 #include <algorithm>
  45 #include <functional>
  46
  47 #include "array_delete.h"
  48 #include "BasicThread.h"
  49 #include "node_manager_impl.h"
  50 #include "functional_tools.h"
  51
  52 using namespace std;
  53
  54 __USE_CORTEX_NAMESPACE
  55 #define D_METHOD(x) //PRINT (x)
  56 #define D_ROSTER(x) //PRINT (x)
  57 #define D_LOCK(x) //PRINT (x)
  58
  59
  60
  61 // -------------------------------------------------------- //
  62 // *** ctor/dtor
  63 // -------------------------------------------------------- //
  64
  65 // free the group, including all nodes within it
  66 // (this call will result in the eventual deletion of the object.)
  67 // returns B_OK on success; B_NOT_ALLOWED if release() has
  68 // already been called; other error codes if the Media Roster
  69 // call fails.
  70 // * THE MANAGER MUST BE LOCKED
  71
  72 status_t NodeGroup::release() {
  73
  74         D_METHOD((
  75                 "NodeGroup::release()\n"));
  76
  77         if(isReleased())
  78                 return B_NOT_ALLOWED;
  79
  80         // clean up
  81         lock();
  82
  83         // halt all nodes
  84         _stop();
  85
  86         // remove & release all nodes
  87         // +++++ causes triply-nested lock: eww!
  88         while(m_nodes.size()) {
  89                 NodeRef* last = m_nodes.back();
  90                 removeNode(m_nodes.size()-1);
  91                 last->release();
  92         }
  93
  94         unlock();
  95
  96 // [e.moon 7nov99]
  97 // removing the released group is now NodeManager's responsibility
  98 //
  99         // remove from NodeManager
 100         if(!m_manager->lock()) {
 101                 ASSERT(!"* m_manager->lock() failed.\n");
 102         }
 103         m_manager->_removeGroup(this);
 104         m_manager->unlock();
 105
 106         // hand off to IObservable
 107         return _inherited::release();
 108 }
 109
 110 // call release() rather than deleting NodeGroup objects
 111 NodeGroup::~NodeGroup() {
 112
 113         Autolock _l(this);
 114         D_METHOD((
 115                 "~NodeGroup()\n"));
 116
 117         ASSERT(!m_nodes.size());
 118
 119         if(m_timeSourceObj) {
 120                 m_timeSourceObj->Release();
 121                 m_timeSourceObj = 0;
 122         }
 123 }
 124
 125
 126 // -------------------------------------------------------- //
 127 // *** accessors
 128 // -------------------------------------------------------- //
 129
 130 // [e.moon 13oct99] moved to header
 131 //inline uint32 NodeGroup::id() const { return m_id; }
 132
 133 // -------------------------------------------------------- //
 134 // *** operations
 135 // -------------------------------------------------------- //
 136
 137 // name access
 138 const char* NodeGroup::name() const {
 139         Autolock _l(this);
 140         return m_name.String();
 141 }
 142
 143 status_t NodeGroup::setName(const char* name) {
 144         Autolock _l(this);
 145         m_name = name;
 146         return B_OK;
 147 }
 148
 149 // content access
 150 uint32 NodeGroup::countNodes() const {
 151         Autolock _l(this);
 152         return m_nodes.size();
 153 }
 154
 155 NodeRef* NodeGroup::nodeAt(
 156         uint32                                                                                  index) const {
 157         Autolock _l(this);
 158         return (index < m_nodes.size()) ?
 159                 m_nodes[index] :
 160                 0;
 161 }
 162
 163 // add/remove nodes:
 164 // - you may only add a node with no current group.
 165 // - nodes added during playback will be started;
 166 //   nodes removed during playback will be stopped (unless
 167 //   the NO_START_STOP transport restriction flag is set
 168 //   for a given node.)
 169
 170 status_t NodeGroup::addNode(
 171         NodeRef*                                                                                node) {
 172
 173         D_METHOD((
 174                 "NodeGroup::addNode()\n"));
 175
 176         // lock the manager first; if the node has no current group,
 177         // this locks the node.
 178         m_manager->lock();
 179
 180         Autolock _l(this);
 181
 182         // precondition: GROUP_LOCKED not set
 183         if(m_flags & GROUP_LOCKED)
 184                 return B_NOT_ALLOWED;
 185
 186         // precondition: no current group
 187         if(node->m_group) {
 188                 // [e.moon 28sep99] whoops, forgot one
 189                 PRINT((
 190                         "!!! node already in group '%s'\n", node->m_group->name()));
 191
 192                 m_manager->unlock();
 193                 return B_NOT_ALLOWED;
 194         }
 195
 196         // add it
 197         m_nodes.push_back(node);
 198         node->_setGroup(this);
 199
 200         // release the manager
 201         m_manager->unlock();
 202
 203         // first node? the transport is now ready to start
 204         if(m_nodes.size() == 1) {
 205                 _changeState(TRANSPORT_INVALID, TRANSPORT_STOPPED);
 206         }
 207 //
 208 //      if(m_syncNode == media_node::null) {
 209 //              // assign as sync node
 210 //              setSyncNode(node->node());
 211 //      }
 212 //
 213         // initialize the new node
 214         status_t err = node->_initTransportState();
 215         if(err < B_OK)
 216                 return err;
 217
 218         // set time source
 219         node->_setTimeSource(m_timeSource.node);
 220
 221         // set run mode
 222         node->_setRunMode(m_runMode);
 223
 224         // add to cycle set if need be
 225         // +++++ should I call _cycleAddRef() instead?
 226         if(node->m_cycle)
 227                 _refCycleChanged(node);
 228
 229         if(m_transportState == TRANSPORT_RUNNING) {
 230                 // +++++ start if necessary!
 231         }
 232         // +++++ not started if TRANSPORT_ROLLING: is that proper? [e.moon 11oct99]
 233
 234         // send notification
 235         if(!LockLooper()) {
 236                 ASSERT(!"LockLooper() failed.");
 237         }
 238         BMessage m(M_NODE_ADDED);
 239         m.AddInt32("groupID", id());
 240         m.AddInt32("nodeID", node->id());
 241         notify(&m);
 242         UnlockLooper();
 243
 244         // success
 245         return B_OK;
 246 }
 247
 248
 249 status_t NodeGroup::removeNode(
 250         NodeRef*                                                                                node) {
 251
 252         D_METHOD((
 253                 "NodeGroup::removeNode()\n"));
 254
 255         // lock the manager first; once the node is ungrouped,
 256         // the manager lock applies to it
 257         m_manager->lock();
 258
 259         Autolock _l(this);
 260
 261         // precondition: this must be the node's group
 262         if(node->m_group != this) {
 263                 // [e.moon 28sep99] whoops, forgot one
 264                 PRINT((
 265                         "!!! node not in group '%s'\n", node->m_group->name()));
 266
 267                 m_manager->unlock();
 268                 return B_NOT_ALLOWED;
 269         }
 270
 271         // remove from the cycle set
 272         if(node->m_cycle)
 273                 _cycleRemoveRef(node);
 274
 275         // remove it
 276         ASSERT(m_nodes.size());
 277         remove(
 278                 m_nodes.begin(),
 279                 m_nodes.end(),
 280                 node);
 281
 282         // should have removed one and only one entry
 283         m_nodes.resize(m_nodes.size()-1);
 284
 285 //      // 6aug99: the timesource is now the sync node...
 286 //      // is this the sync node? reassign if so
 287 //
 288 //      if(node->node() == m_syncNode) {
 289 //
 290 //              // look for another sync-capable node
 291 //              bool found = false;
 292 //              for(int n = 0; !found && n < m_nodes.size(); ++n)
 293 //                      if(setSyncNode(m_nodes[n]->node()) == B_OK)
 294 //                              found = true;
 295 //
 296 //              // no luck? admit defeat:
 297 //              if(!found) {
 298 //                      PRINT((
 299 //                              "* NodeGroup::removeNode(): no sync-capable nodes left!\n"));
 300 //
 301 //                      // +++++ stop & set to invalid state?
 302 //
 303 //                      setSyncNode(media_node::null);
 304 //              }
 305 //      }
 306
 307         // stop the node if necessary
 308         status_t err = node->_stop();
 309         if(err < B_OK) {
 310                 PRINT((
 311                         "*** NodeGroup::removeNode('%s'): error from node->_stop():\n"
 312                         "    %s\n",
 313                         node->name(),
 314                         strerror(err)));
 315         }
 316
 317         // clear the node's group pointer
 318         node->_setGroup(0);
 319
 320         // release the manager lock; the node is now ungrouped and
 321         // unlocked
 322         m_manager->unlock();
 323
 324         // was that the last node? stop/disable the transport if so
 325         if(!m_nodes.size()) {
 326
 327                 // +++++ kill sync thread(s)
 328
 329                 _changeState(TRANSPORT_INVALID);
 330         }
 331
 332         // send notification
 333         if(!LockLooper()) {
 334                 ASSERT(!"LockLooper() failed.");
 335         }
 336         BMessage m(M_NODE_REMOVED);
 337         m.AddInt32("groupID", id());
 338         m.AddInt32("nodeID", node->id());
 339         notify(&m);
 340         UnlockLooper();
 341
 342         // success
 343         return B_OK;
 344 }
 345
 346 status_t NodeGroup::removeNode(
 347         uint32                                                                                  index) {
 348
 349         D_METHOD((
 350                 "NodeGroup::removeNode(by index)\n"));
 351
 352         // +++++ icky nested lock
 353         Autolock _l(this);
 354
 355         ASSERT(m_nodes.size() > index);
 356         return removeNode(m_nodes[index]);
 357 }
 358
 359 uint32 NodeGroup::groupFlags() const {
 360         Autolock _l(this);
 361         return m_flags;
 362 }
 363
 364 status_t NodeGroup::setGroupFlags(
 365         uint32                                                                                  flags) {
 366         Autolock _l(this);
 367         m_flags = flags;
 368         return B_OK;
 369 }
 370
 371
 372 // returns true if one or more nodes in the group have cycling
 373 // enabled, and the start- and end-positions are valid
 374 bool NodeGroup::canCycle() const {
 375         Autolock _l(this);
 376
 377         return
 378                 m_cycleNodes.size() > 0 &&
 379                 m_endPosition - m_startPosition > s_minCyclePeriod;
 380 }
 381
 382 // -------------------------------------------------------- //
 383 // *** TRANSPORT POSITIONING (LOCK REQUIRED)
 384 // -------------------------------------------------------- //
 385
 386 // Fetch the current transport state
 387
 388 NodeGroup::transport_state_t NodeGroup::transportState() const {
 389         Autolock _l(this);
 390         return m_transportState;
 391 }
 392
 393 // Set the starting media time:
 394 //   This is the point at which playback will begin in any media
 395 //   files/documents being played by the nodes in this group.
 396 //   When cycle mode is enabled, this is the point to which each
 397 //   node will be seek'd at the end of each cycle (loop).
 398 //
 399 //   The starting time can't be changed in the B_OFFLINE run mode
 400 //   (this call will return an error.)
 401
 402 status_t NodeGroup::setStartPosition(
 403         bigtime_t                                                                               start) {
 404         Autolock _l(this);
 405
 406         D_METHOD((
 407                 "NodeGroup::setStartPosition(%Ld)\n", start));
 408
 409         if(
 410                 m_transportState == TRANSPORT_RUNNING ||
 411                 m_transportState == TRANSPORT_ROLLING ||
 412                 m_transportState == TRANSPORT_STARTING) {
 413
 414                 if(m_runMode == BMediaNode::B_OFFLINE)
 415                         return B_NOT_ALLOWED;
 416
 417                 ASSERT(m_timeSourceObj);
 418
 419                 if(_cycleValid()) {
 420                         if(m_timeSourceObj->Now() >= m_cycleDeadline) {
 421                                 // too late to change start position; defer
 422 //                              PRINT((" - deferred\n"));
 423                                 m_newStartPosition = start;
 424                                 m_newStart = true;
 425                                 return B_OK;
 426                         }
 427
 428                         // not at deadline yet; fall through to set start position
 429                 }
 430         }
 431
 432         m_startPosition = start;
 433
 434         // +++++ notify [e.moon 11oct99]
 435
 436         return B_OK;
 437 }
 438
 439 // Fetch the starting position:
 440
 441 // +++++ if a previously-set start position was deferred, it won't be
 442 //       returned yet
 443
 444 bigtime_t NodeGroup::startPosition() const {
 445         Autolock _l(this);
 446
 447         return m_startPosition;
 448 }
 449
 450 // Set the ending media time:
 451 //   This is the point at which playback will end relative to
 452 //   media documents begin played by the nodes in this group;
 453 //   in cycle mode, this specifies the loop point.  If the
 454 //   ending time is less than or equal to the starting time,
 455 //   the transport will continue until stopped manually.
 456 //   If the end position is changed while the transport is playing,
 457 //   it must take effect retroactively (if it's before the current
 458 //   position and looping is enabled, all nodes must 'warp' to
 459 //   the proper post-loop position.)
 460 //
 461 //   The ending time can't be changed if run mode is B_OFFLINE and
 462 //   the transport is running (this call will return an error.)
 463
 464 status_t NodeGroup::setEndPosition(
 465         bigtime_t                                                                               end) {
 466         Autolock _l(this);
 467
 468         D_METHOD((
 469                 "NodeGroup::setEndPosition(%Ld)\n", end));
 470
 471         if(
 472                 m_transportState == TRANSPORT_RUNNING ||
 473                 m_transportState == TRANSPORT_ROLLING ||
 474                 m_transportState == TRANSPORT_STARTING) {
 475
 476                 if(m_runMode == BMediaNode::B_OFFLINE)
 477                         return B_NOT_ALLOWED;
 478
 479                 ASSERT(m_timeSourceObj);
 480
 481                 bigtime_t endDelta = end - m_endPosition;
 482
 483                 if(_cycleValid()) {
 484                         if(m_timeSourceObj->Now() >= m_cycleDeadline + endDelta) {
 485                                 // too late to change end position; defer
 486 //                              PRINT((" - deferred\n"));
 487                                 m_newEndPosition = end;
 488                                 m_newEnd = true;
 489                                 return B_OK;
 490                         }
 491                         else {
 492                                 // set new end position
 493                                 m_endPosition = end;
 494
 495                                 // inform thread
 496                                 ASSERT(m_cyclePort);
 497                                 write_port(
 498                                         m_cyclePort,
 499                                         _CYCLE_END_CHANGED,
 500                                         0,
 501                                         0);
 502 //
 503 //                              // restart nodes' cycle threads with new end position
 504 //                              _cycleInit(m_cycleStart);
 505 //                              for(node_set::iterator it = m_cycleNodes.begin();
 506 //                                      it != m_cycleNodes.end(); ++it) {
 507 //                                      (*it)->_scheduleCycle(m_cycleBoundary);
 508 //                              }
 509 //                              return B_OK;
 510                         }
 511                 }
 512         }
 513
 514         m_endPosition = end;
 515
 516         // +++++ notify [e.moon 11oct99]
 517
 518         return B_OK;
 519 }
 520
 521
 522 // Fetch the end position:
 523 //   Note that if the end position is less than or equal to the start
 524 //   position, it's ignored.
 525
 526 // +++++ if a previously-set end position was deferred, it won't be
 527 //       returned yet
 528
 529 bigtime_t NodeGroup::endPosition() const {
 530         Autolock _l(this);
 531         return m_endPosition;
 532 }
 533
 534 // -------------------------------------------------------- //
 535 // *** TRANSPORT OPERATIONS (LOCK REQUIRED)
 536 // -------------------------------------------------------- //
 537
 538 // Preroll the group:
 539 //   Seeks, then prerolls, each node in the group (honoring the
 540 //   NO_SEEK and NO_PREROLL flags.)  This ensures that the group
 541 //   can start as quickly as possible.
 542 //
 543 //   Does not return until all nodes in the group have been
 544 //   prepared.
 545
 546 status_t NodeGroup::preroll() {
 547         D_METHOD((
 548                 "NodeGroup::preroll()\n"));
 549
 550         Autolock _l(this);
 551         return _preroll();
 552 }
 553
 554 // Start all nodes in the group:
 555 //   Nodes with the NO_START_STOP flag aren't molested.
 556
 557 status_t NodeGroup::start() {
 558         D_METHOD((
 559                 "NodeGroup::start()\n"));
 560
 561         Autolock _l(this);
 562         return _start();
 563 }
 564
 565 // Stop all nodes in the group:
 566 //   Nodes with the NO_START_STOP flag aren't molested.
 567
 568 status_t NodeGroup::stop() {
 569         D_METHOD((
 570                 "NodeGroup::stop()\n"));
 571
 572         Autolock _l(this);
 573         return _stop();
 574 }
 575
 576 // Roll all nodes in the group:
 577 //   Queues a start and stop atomically (via BMediaRoster::RollNode()).
 578 //   Returns B_NOT_ALLOWED if endPosition <= startPosition;
 579
 580 status_t NodeGroup::roll() {
 581         D_METHOD((
 582                 "NodeGroup::roll()\n"));
 583
 584         Autolock _l(this);
 585         return _roll();
 586 }
 587
 588 // -------------------------------------------------------- //
 589 // *** TIME SOURCE & RUN-MODE OPERATIONS (LOCK REQUIRED)
 590 // -------------------------------------------------------- //
 591
 592 // time source control:
 593 //   getTimeSource():
 594 //   returns B_ERROR if no time source has been set; otherwise,
 595 //   returns the node ID of the current time source for all
 596 //   nodes in the group.
 597 //
 598 //   setTimeSource():
 599 //   Calls SetTimeSourceFor() on every node in the group.
 600 //   The group must be stopped; B_NOT_ALLOWED will be returned
 601 //   if the state is TRANSPORT_RUNNING or TRANSPORT_ROLLING.
 602
 603 status_t NodeGroup::getTimeSource(
 604         media_node*                                                                     outTimeSource) const {
 605         Autolock _l(this);
 606
 607         if(m_timeSource != media_node::null) {
 608                 *outTimeSource = m_timeSource;
 609                 return B_OK;
 610         }
 611         return B_ERROR;
 612 }
 613
 614 status_t NodeGroup::setTimeSource(
 615         const media_node&                                               timeSource) {
 616
 617         Autolock _l(this);
 618
 619         if(m_transportState == TRANSPORT_RUNNING || m_transportState == TRANSPORT_ROLLING)
 620                 return B_NOT_ALLOWED;
 621
 622         if(m_timeSourceObj)
 623                 m_timeSourceObj->Release();
 624
 625         m_timeSource = timeSource;
 626
 627         // cache a BTimeSource*
 628         m_timeSourceObj = m_manager->roster->MakeTimeSourceFor(timeSource);
 629         ASSERT(m_timeSourceObj);
 630
 631         // apply new time source to all nodes
 632         for_each(
 633                 m_nodes.begin(),
 634                 m_nodes.end(),
 635                 bind2nd(
 636                         mem_fun(&NodeRef::_setTimeSource),
 637                         m_timeSource.node
 638                 )
 639         );
 640
 641 //      // try to set as sync node
 642 //      err = setSyncNode(timeSource);
 643 //      if(err < B_OK) {
 644 //              PRINT((
 645 //                      "* NodeGroup::setTimeSource(): setSyncNode() failed: %s\n",
 646 //                      strerror(err)));
 647 //      }
 648
 649         // notify
 650         if(!LockLooper()) {
 651                 ASSERT(!"LockLooper() failed.");
 652         }
 653         BMessage m(M_TIME_SOURCE_CHANGED);
 654         m.AddInt32("groupID", id());
 655         m.AddInt32("timeSourceID", timeSource.node);
 656         notify(&m);
 657         UnlockLooper();
 658
 659         return B_OK;
 660 }
 661
 662 // run mode access:
 663 //   Sets the default run mode for the group.  This will be
 664 //   applied to every node with a wildcard (0) run mode.
 665 //
 666 //   Special case: if the run mode is B_OFFLINE, it will be
 667 //   applied to all nodes in the group.
 668
 669 BMediaNode::run_mode NodeGroup::runMode() const {
 670         Autolock _l(this);
 671         return m_runMode;
 672 }
 673
 674 status_t NodeGroup::setRunMode(BMediaNode::run_mode mode) {
 675         Autolock _l(this);
 676
 677         m_runMode = mode;
 678
 679         // apply to all nodes
 680         for_each(
 681                 m_nodes.begin(),
 682                 m_nodes.end(),
 683                 bind2nd(
 684                         mem_fun(&NodeRef::_setRunModeAuto),
 685                         m_runMode
 686                 )
 687 //              bound_method(
 688 //                      *this,
 689 //                      &NodeGroup::setRunModeFor)
 690         );
 691
 692
 693         return B_OK;
 694 }
 695
 696 // -------------------------------------------------------- //
 697 // *** BHandler
 698 // -------------------------------------------------------- //
 699
 700 void NodeGroup::MessageReceived(
 701         BMessage*                                                                               message) {
 702
 703 //      PRINT((
 704 //              "NodeGroup::MessageReceived():\n"));
 705 //      message->PrintToStream();
 706         status_t err;
 707
 708         switch(message->what) {
 709                 case M_SET_TIME_SOURCE:
 710                         {
 711                                 media_node timeSource;
 712                                 void* data;
 713                                 ssize_t dataSize;
 714                                 err = message->FindData(
 715                                         "timeSourceNode",
 716                                         B_RAW_TYPE,
 717                                         (const void**)&data,
 718                                         &dataSize);
 719                                 if(err < B_OK) {
 720                                         PRINT((
 721                                                 "* NodeGroup::MessageReceived(M_SET_TIME_SOURCE):\n"
 722                                                 "  no timeSourceNode!\n"));
 723                                         break;
 724                                 }
 725                                 timeSource = *(media_node*)data;
 726
 727                                 setTimeSource(timeSource);
 728                         }
 729                         break;
 730
 731                 case M_SET_RUN_MODE:
 732                         {
 733                                 uint32 runMode;
 734                                 err = message->FindInt32("runMode", (int32*)&runMode);
 735                                 if(err < B_OK) {
 736                                         PRINT((
 737                                                 "* NodeGroup::MessageReceived(M_SET_RUN_MODE):\n"
 738                                                 "  no runMode!\n"));
 739                                         break;
 740                                 }
 741
 742                                 if(runMode < BMediaNode::B_OFFLINE ||
 743                                         runMode > BMediaNode::B_RECORDING) {
 744                                         PRINT((
 745                                                 "* NodeGroup::MessageReceived(M_SET_RUN_MODE):\n"
 746                                                 "  invalid run mode (%" B_PRIu32 ")\n", runMode));
 747                                         break;
 748                                 }
 749
 750                                 setRunMode((BMediaNode::run_mode)runMode);
 751                         }
 752                         break;
 753
 754                 case M_SET_START_POSITION:
 755                         {
 756                                 bigtime_t position;
 757                                 err = message->FindInt64("position", (int64*)&position);
 758                                 if(err < B_OK) {
 759                                         PRINT((
 760                                                 "* NodeGroup::MessageReceived(M_SET_START_POSITION):\n"
 761                                                 "  no position!\n"));
 762                                         break;
 763                                 }
 764                                 setStartPosition(position);
 765                         }
 766                         break;
 767
 768                 case M_SET_END_POSITION:
 769                         {
 770                                 bigtime_t position;
 771                                 err = message->FindInt64("position", (int64*)&position);
 772                                 if(err < B_OK) {
 773                                         PRINT((
 774                                                 "* NodeGroup::MessageReceived(M_SET_END_POSITION):\n"
 775                                                 "  no position!\n"));
 776                                         break;
 777                                 }
 778                                 setEndPosition(position);
 779                         }
 780                         break;
 781
 782                 case M_PREROLL:
 783                         preroll();
 784                         break;
 785
 786                 case M_START:
 787                         start();
 788                         break;
 789
 790                 case M_STOP:
 791                         stop();
 792                         break;
 793
 794                 case M_ROLL:
 795                         roll();
 796                         break;
 797
 798                 default:
 799                         _inherited::MessageReceived(message);
 800                         break;
 801         }
 802 }
 803
 804
 805 // -------------------------------------------------------- //
 806 // *** IPersistent
 807 // -------------------------------------------------------- //
 808
 809 // !
 810 #if CORTEX_XML
 811 // !
 812
 813 // +++++
 814
 815 // Default constructor
 816 NodeGroup::NodeGroup() :
 817         m_manager(0) {} // +++++ finish initialization
 818
 819
 820 // !
 821 #endif /*CORTEX_XML*/
 822 // !
 823
 824 // -------------------------------------------------------- //
 825 // *** IObservable:             [19aug99]
 826 // -------------------------------------------------------- //
 827
 828 void NodeGroup::observerAdded(
 829         const BMessenger&                               observer) {
 830
 831         BMessage m(M_OBSERVER_ADDED);
 832         m.AddInt32("groupID", id());
 833         m.AddMessenger("target", BMessenger(this));
 834         observer.SendMessage(&m);
 835 }
 836
 837 void NodeGroup::observerRemoved(
 838         const BMessenger&                               observer) {
 839
 840         BMessage m(M_OBSERVER_REMOVED);
 841         m.AddInt32("groupID", id());
 842         m.AddMessenger("target", BMessenger(this));
 843         observer.SendMessage(&m);
 844 }
 845
 846 void NodeGroup::notifyRelease() {
 847
 848         BMessage m(M_RELEASED);
 849         m.AddInt32("groupID", id());
 850         m.AddMessenger("target", BMessenger(this));
 851         notify(&m);
 852 }
 853
 854 void NodeGroup::releaseComplete() {
 855         // +++++
 856 }
 857
 858 // -------------------------------------------------------- //
 859 // *** ILockable: pass lock requests to m_lock
 860 // -------------------------------------------------------- //
 861
 862 bool NodeGroup::lock(
 863         lock_t type,
 864         bigtime_t timeout) {
 865
 866         D_LOCK(("*** NodeGroup::lock(): %ld\n", find_thread(0)));
 867
 868         ASSERT(type == WRITE);
 869         status_t err = m_lock.LockWithTimeout(timeout);
 870
 871         D_LOCK(("*** NodeGroup::lock() ACQUIRED: %ld\n", find_thread(0)));
 872
 873         return err == B_OK;
 874 }
 875
 876 bool NodeGroup::unlock(
 877         lock_t type) {
 878
 879         D_LOCK(("*** NodeGroup::unlock(): %ld\n", find_thread(0)));
 880
 881         ASSERT(type == WRITE);
 882         m_lock.Unlock();
 883
 884         D_LOCK(("*** NodeGroup::unlock() RELEASED: %ld\n", find_thread(0)));
 885
 886         return true;
 887 }
 888
 889 bool NodeGroup::isLocked(
 890         lock_t type) const {
 891
 892         ASSERT(type == WRITE);
 893         return m_lock.IsLocked();
 894 }
 895
 896 // -------------------------------------------------------- //
 897 // *** ctor (accessible to NodeManager)
 898 // -------------------------------------------------------- //
 899
 900 NodeGroup::NodeGroup(
 901         const char*                                                                     name,
 902         NodeManager*                                                            manager,
 903         BMediaNode::run_mode                            runMode) :
 904
 905         ObservableHandler(name),
 906         m_lock("NodeGroup::m_lock"),
 907         m_manager(manager),
 908         m_id(NextID()),
 909         m_name(name),
 910         m_flags(0),
 911         m_transportState(TRANSPORT_INVALID),
 912         m_runMode(runMode),
 913         m_timeSourceObj(0),
 914         m_released(false),
 915         m_cycleThread(0),
 916         m_cyclePort(0),
 917         m_startPosition(0LL),
 918         m_endPosition(0LL),
 919         m_newStart(false),
 920         m_newEnd(false) {
 921
 922         ASSERT(m_manager);
 923
 924         if(!m_manager->Lock()) {
 925                 ASSERT(!"m_manager->Lock() failed");
 926         }
 927         m_manager->AddHandler(this);
 928         m_manager->Unlock();
 929
 930         // set default time source
 931         media_node ts;
 932         D_ROSTER(("# roster->GetTimeSource()\n"));
 933         status_t err = m_manager->roster->GetTimeSource(&ts);
 934         if(err < B_OK) {
 935                 PRINT((
 936                         "*** NodeGroup(): roster->GetTimeSource() failed:\n"
 937                         "    %s\n", strerror(err)));
 938         }
 939         setTimeSource(ts);
 940 }
 941
 942 // -------------------------------------------------------- //
 943 // *** internal operations
 944 // -------------------------------------------------------- //
 945
 946 uint32 NodeGroup::s_nextID = 1;
 947 uint32 NodeGroup::NextID() {
 948         return atomic_add((int32*)&s_nextID, 1);
 949 }
 950
 951 // -------------------------------------------------------- //
 952 // *** ref->group communication (LOCK REQUIRED)
 953 // -------------------------------------------------------- //
 954
 955 // when a NodeRef's cycle state (ie. looping or not looping)
 956 // changes, it must pass that information on via this method
 957
 958 void NodeGroup::_refCycleChanged(
 959         NodeRef*                                                                                ref) {
 960         assert_locked(this);
 961         D_METHOD((
 962                 "NodeGroup::_refCycleChanged('%s')\n",
 963                 ref->name()));
 964
 965         if(ref->m_cycle) {
 966                 _cycleAddRef(ref);
 967         }       else {
 968                 _cycleRemoveRef(ref);
 969         }
 970
 971         // +++++ if running & cycle valid, the node should be properly
 972         //       seek'd and start'd
 973 }
 974
 975
 976 // when a cycling node's latency changes, call this method.
 977
 978 void NodeGroup::_refLatencyChanged(
 979         NodeRef*                                                                                ref) {
 980         assert_locked(this);
 981         D_METHOD((
 982                 "NodeGroup::_refLatencyChanged('%s')\n",
 983                 ref->name()));
 984
 985         if(!_cycleValid())
 986                 return;
 987
 988         // remove & replace ref (positions it properly)
 989         _cycleRemoveRef(ref);
 990         _cycleAddRef(ref);
 991
 992         // slap my thread up
 993         ASSERT(m_cyclePort);
 994         write_port(
 995                 m_cyclePort,
 996                 _CYCLE_LATENCY_CHANGED,
 997                 0,
 998                 0);
 999
1000         // +++++ zat it?
1001 }
1002
1003 // when a NodeRef receives notification that it has been stopped,
1004 // but is labeled as still running, it must call this method.
1005 // [e.moon 11oct99: roll/B_OFFLINE support]
1006
1007 void NodeGroup::_refStopped(
1008         NodeRef*                                                                                ref) {
1009         assert_locked(this);
1010         D_METHOD((
1011                 "NodeGroup::_refStopped('%s')\n",
1012                 ref->name()));
1013
1014         // roll/B_OFFLINE support [e.moon 11oct99]
1015         // (check to see if any other nodes in the group are still running;
1016         //  mark group stopped if not.)
1017         if(m_transportState == TRANSPORT_ROLLING) {
1018                 bool nodesRunning = false;
1019                 for(node_set::iterator it = m_nodes.begin();
1020                         it != m_nodes.end(); ++it) {
1021                         if((*it)->isRunning()) {
1022                                 nodesRunning = true;
1023                                 break;
1024                         }
1025                 }
1026                 if(!nodesRunning)
1027                         // the group has stopped; update transport state
1028                         _changeState(TRANSPORT_STOPPED);
1029
1030         }
1031
1032 }
1033
1034
1035 // -------------------------------------------------------- //
1036 // *** transport helpers (LOCK REQUIRED)
1037 // -------------------------------------------------------- //
1038
1039
1040 // Preroll all nodes in the group; this is the implementation
1041 // of preroll().
1042 // *** this method should not be called from the transport thread
1043 // (since preroll operations can block for a relatively long time.)
1044
1045 status_t NodeGroup::_preroll() {
1046         assert_locked(this);
1047
1048         D_METHOD((
1049                 "NodeGroup::_preroll()\n"));
1050
1051         if(
1052                 m_transportState == TRANSPORT_RUNNING ||
1053                 m_transportState == TRANSPORT_ROLLING)
1054                 // too late
1055                 return B_NOT_ALLOWED;
1056
1057         // * preroll all nodes to the start position
1058
1059         // +++++ currently, if an error is encountered it's ignored.
1060         //       should the whole operation fail if one node couldn't
1061         //       be prerolled?
1062         //
1063         //       My gut response is 'no', since the preroll step is
1064         //       optional, but the caller should have some inkling that
1065         //       one of its nodes didn't behave.
1066
1067 // [e.moon 13oct99] making PPC compiler happy
1068 //      for_each(
1069 //              m_nodes.begin(),
1070 //              m_nodes.end(),
1071 //              bind2nd(
1072 //                      mem_fun(&NodeRef::_preroll),
1073 //                      m_startPosition
1074 //              )
1075 //      );
1076         for(node_set::iterator it = m_nodes.begin();
1077                 it != m_nodes.end(); ++it) {
1078                 (*it)->_preroll(m_startPosition);
1079         }
1080
1081 //    replaces
1082 //              bind2nd(
1083 //                      bound_method(*this, &NodeGroup::prerollNode),
1084 //                      m_startPosition
1085 //              )
1086
1087         return B_OK;
1088 }
1089
1090
1091 //// functor: calculates latency of each node it's handed, caching
1092 //// the largest one found; includes initial latency if nodes report it.
1093 //
1094 //class NodeGroup::calcLatencyFn { public:
1095 //      bigtime_t& maxLatency;
1096 //
1097 //      calcLatencyFn(bigtime_t& _m) : maxLatency(_m) {}
1098 //
1099 //      void operator()(NodeRef* r) {
1100 //              ASSERT(r);
1101 //
1102 ////            PRINT((
1103 ////                    "# calcLatencyFn(): '%s'\n",
1104 ////                    r->name()));
1105 //
1106 //              if(!(r->node().kind & B_BUFFER_PRODUCER)) {
1107 //                      // node can't incur latency
1108 ////                    PRINT((
1109 ////                            "-   not a producer\n"));
1110 //                      return;
1111 //              }
1112 //
1113 //              bigtime_t latency;
1114 //              status_t err =
1115 //                      BMediaRoster::Roster()->GetLatencyFor(
1116 //                              r->node(),
1117 //                              &latency);
1118 //              if(err < B_OK) {
1119 //                      PRINT((
1120 //                              "* calcLatencyFn: GetLatencyFor() failed: %s\n",
1121 //                              strerror(err)));
1122 //                      return;
1123 //              }
1124 ////            PRINT(("-   %Ld\n", latency));
1125 //
1126 //              bigtime_t add;
1127 //              err = BMediaRoster::Roster()->GetInitialLatencyFor(
1128 //                      r->node(),
1129 //                      &add);
1130 ////            PRINT(("-   %Ld\n", add));
1131 //              if(err < B_OK) {
1132 //                      PRINT((
1133 //                              "* calcLatencyFn: GetInitialLatencyFor() failed: %s\n",
1134 //                              strerror(err)));
1135 //              }
1136 //              else
1137 //                      latency += add;
1138 //
1139 //              if(latency > maxLatency)
1140 //                      maxLatency = latency;
1141 //
1142 ////            PRINT((
1143 ////                    "-   max latency: %Ld\n",
1144 ////                    maxLatency));
1145 //      }
1146 //};
1147
1148 // Start all nodes in the group; this is the implementation of
1149 // start().  Fails if the run mode is B_OFFLINE; use _roll() instead
1150 // in that case.
1151 //
1152 // (this may be called from the transport thread or from
1153 //  an API-implementation method.)
1154
1155 status_t NodeGroup::_start() {
1156         assert_locked(this);
1157
1158         D_METHOD((
1159                 "NodeGroup::_start()\n"));
1160         status_t err;
1161
1162         if(m_transportState != TRANSPORT_STOPPED)
1163                 return B_NOT_ALLOWED;
1164
1165         if(m_runMode == BMediaNode::B_OFFLINE)
1166                 return B_NOT_ALLOWED;
1167
1168         ASSERT(m_nodes.size());
1169
1170         _changeState(TRANSPORT_STARTING);
1171
1172         // * Find the highest latency in the group
1173
1174         bigtime_t offset = 0LL;
1175         calcLatencyFn _f(offset);
1176         for_each(
1177                 m_nodes.begin(),
1178                 m_nodes.end(),
1179                 _f);
1180
1181         offset += s_rosterLatency;
1182         PRINT((
1183                 "- offset: %" B_PRIdBIGTIME "\n", offset));
1184
1185         // * Seek all nodes (in case one or more failed to preroll)
1186
1187         for(node_set::iterator it = m_nodes.begin();
1188                 it != m_nodes.end(); ++it) {
1189                 err = (*it)->_seekStopped(m_startPosition);
1190                 if(err < B_OK) {
1191                         PRINT((
1192                                 "! NodeGroup('%s')::_start():\n"
1193                                 "  ref('%s')->_seekStopped(%" B_PRIdBIGTIME ") failed:\n"
1194                                 "  %s\n",
1195                                 name(), (*it)->name(), m_startPosition,
1196                                 strerror(err)));
1197
1198                         // +++++ continue?
1199                 }
1200         }
1201
1202         // * Start all nodes, allowing for the max latency found
1203
1204         ASSERT(m_timeSourceObj);
1205         bigtime_t when = m_timeSourceObj->Now() + offset;
1206
1207         // 10aug99: initialize cycle (loop) settings
1208         if(_cycleValid()) {
1209                 _initCycleThread();
1210                 _cycleInit(when);
1211         }
1212
1213         // start the nodes
1214         for(node_set::iterator it = m_nodes.begin();
1215                 it != m_nodes.end(); ++it) {
1216                 err = (*it)->_start(when);
1217                 if(err < B_OK) {
1218                         PRINT((
1219                                 "! NodeGroup('%s')::_start():\n"
1220                                 "  ref('%s')->_start(%" B_PRIdBIGTIME ") failed:\n"
1221                                 "  %s\n",
1222                                 name(), (*it)->name(), when,
1223                                 strerror(err)));
1224
1225                         // +++++ continue?
1226                 }
1227         }
1228
1229         // notify observers
1230         _changeState(TRANSPORT_RUNNING);
1231         return B_OK;
1232 }
1233
1234 // Stop all nodes in the group; this is the implementation of
1235 // stop().
1236 //
1237 // (this may be called from the transport thread or from
1238 //  an API-implementation method.)
1239
1240 status_t NodeGroup::_stop() {
1241
1242         D_METHOD((
1243                 "NodeGroup::_stop()\n"));
1244
1245         assert_locked(this);
1246
1247         if(
1248                 m_transportState != TRANSPORT_RUNNING &&
1249                 m_transportState != TRANSPORT_ROLLING)
1250                 return B_NOT_ALLOWED;
1251
1252         _changeState(TRANSPORT_STOPPING);
1253
1254         // * stop the cycle thread if need be
1255         _destroyCycleThread();
1256
1257         // * stop all nodes
1258         //   +++++ error reports would be nice
1259
1260         for_each(
1261                 m_nodes.begin(),
1262                 m_nodes.end(),
1263                 mem_fun(&NodeRef::_stop)
1264         );
1265
1266         // update transport state
1267         _changeState(TRANSPORT_STOPPED);
1268
1269         return B_OK;
1270 }
1271
1272 // Roll all nodes in the group; this is the implementation of
1273 // roll().
1274 //
1275 // (this may be called from the transport thread or from
1276 //  an API-implementation method.)
1277
1278 status_t NodeGroup::_roll() {
1279
1280         D_METHOD((
1281                 "NodeGroup::_roll()\n"));
1282         assert_locked(this);
1283         status_t err;
1284
1285         if(m_transportState != TRANSPORT_STOPPED)
1286                 return B_NOT_ALLOWED;
1287
1288         bigtime_t period = m_endPosition - m_startPosition;
1289         if(period <= 0LL)
1290                 return B_NOT_ALLOWED;
1291
1292         _changeState(TRANSPORT_STARTING);
1293
1294         bigtime_t tpStart = 0LL;
1295         bigtime_t tpStop = period;
1296
1297         if(m_runMode != BMediaNode::B_OFFLINE) {
1298
1299                 // * Find the highest latency in the group
1300                 bigtime_t offset = 0LL;
1301                 calcLatencyFn _f(offset);
1302                 for_each(
1303                         m_nodes.begin(),
1304                         m_nodes.end(),
1305                         _f);
1306
1307                 offset += s_rosterLatency;
1308                 PRINT((
1309                         "- offset: %" B_PRIdBIGTIME "\n", offset));
1310
1311                 ASSERT(m_timeSourceObj);
1312                 tpStart = m_timeSourceObj->Now() + offset;
1313                 tpStop += tpStart;
1314         }
1315
1316         // * Roll all nodes; watch for errors
1317         bool allFailed = true;
1318         err = B_OK;
1319         for(
1320                 node_set::iterator it = m_nodes.begin();
1321                 it != m_nodes.end(); ++it) {
1322
1323                 status_t e = (*it)->_roll(
1324                         tpStart,
1325                         tpStop,
1326                         m_startPosition);
1327                 if(e < B_OK)
1328                         err = e;
1329                 else
1330                         allFailed = false;
1331         }
1332
1333         if(!allFailed)
1334                 // notify observers
1335                 _changeState(TRANSPORT_ROLLING);
1336
1337         return err;
1338 }
1339
1340
1341 // State transition; notify listeners
1342 // +++++ [18aug99] DANGER: should notification happen in the middle
1343 //                         of such an operation?
1344 inline void NodeGroup::_changeState(
1345         transport_state_t                       to) {
1346
1347         assert_locked(this);
1348
1349         m_transportState = to;
1350
1351         if(!LockLooper()) {
1352                 ASSERT(!"LockLooper() failed.");
1353         }
1354         BMessage m(M_TRANSPORT_STATE_CHANGED);
1355         m.AddInt32("groupID", id());
1356         m.AddInt32("transportState", m_transportState);
1357         notify(&m);
1358         UnlockLooper();
1359 }
1360
1361 // Enforce a state transition, and notify listeners
1362 inline void NodeGroup::_changeState(
1363         transport_state_t                       from,
1364         transport_state_t                       to) {
1365
1366         assert_locked(this);
1367         ASSERT(m_transportState == from);
1368
1369         _changeState(to);
1370 }
1371
1372
1373 // -------------------------------------------------------- //
1374 // *** cycle thread & helpers (LOCK REQUIRED)
1375 // -------------------------------------------------------- //
1376
1377 // *** cycle port definitions
1378
1379 const int32                                     _portLength                     = 32;
1380 const char* const               _portName                               = "NodeGroup::m_cyclePort";
1381 const size_t                            _portMsgMaxSize = 256;
1382
1383
1384 // set up the cycle thread (including its kernel port)
1385 status_t NodeGroup::_initCycleThread() {
1386         assert_locked(this);
1387         status_t err;
1388         D_METHOD((
1389                 "NodeGroup::_initCycleThread()\n"));
1390
1391         if(m_cycleThread) {
1392                 // thread is still alive
1393                 err = _destroyCycleThread();
1394                 if(err < B_OK)
1395                         return err;
1396         }
1397
1398         // create
1399         m_cycleThreadDone = false;
1400         m_cycleThread = spawn_thread(
1401                 &_CycleThread,
1402                 "NodeGroup[cycleThread]",
1403                 B_NORMAL_PRIORITY,
1404                 (void*)this);
1405         if(m_cycleThread < B_OK) {
1406                 PRINT((
1407                         "* NodeGroup::_initCycleThread(): spawn_thread() failed:\n"
1408                         "  %s\n",
1409                         strerror(m_cycleThread)));
1410                 return m_cycleThread;
1411         }
1412
1413         // launch
1414         return resume_thread(m_cycleThread);
1415 }
1416
1417 // shut down the cycle thread/port
1418 status_t NodeGroup::_destroyCycleThread() {
1419         assert_locked(this);
1420         status_t err;
1421         D_METHOD((
1422                 "NodeGroup::_destroyCycleThread()\n"));
1423
1424         if(!m_cycleThread)
1425                 return B_OK;
1426
1427         if(!m_cycleThreadDone) {
1428                 // kill the thread
1429                 ASSERT(m_cyclePort);
1430                 err = write_port_etc(
1431                         m_cyclePort,
1432                         _CYCLE_STOP,
1433                         0,
1434                         0,
1435                         B_TIMEOUT,
1436                         10000LL);
1437
1438                 if(err < B_OK) {
1439                         // bad thread.  die, thread, die.
1440                         PRINT((
1441                                 "* NodeGroup::_destroyCycleThread(): port write failed; killing.\n"));
1442                         delete_port(m_cyclePort);
1443                         m_cyclePort = 0;
1444                         kill_thread(m_cycleThread);
1445                         m_cycleThread = 0;
1446                         return B_OK;
1447                 }
1448
1449                 // the thread got the message; wait for it to quit
1450                 unlock();
1451                 while(wait_for_thread(m_cycleThread, &err) == B_INTERRUPTED) {
1452                         PRINT((
1453                                 "! wait_for_thread(m_cycleThread, &err) == B_INTERRUPTED\n"));
1454                 }
1455                 lock();
1456         }
1457
1458         // it's up to the thread to close its port
1459         ASSERT(!m_cyclePort);
1460
1461         m_cycleThread = 0;
1462
1463         return B_OK;
1464 }
1465
1466
1467 // 1) do the current positions specify a valid cycle region?
1468 // 2) are any nodes in the group cycle-enabled?
1469
1470 bool NodeGroup::_cycleValid() {
1471         assert_locked(this);
1472         return
1473                 (m_transportState == TRANSPORT_RUNNING ||
1474                  m_transportState == TRANSPORT_STARTING) &&
1475                  canCycle();
1476 }
1477
1478 // initialize the cycle members (call when starting)
1479
1480 void NodeGroup::_cycleInit(
1481         bigtime_t                                                                               startTime) {
1482         assert_locked(this);
1483         ASSERT(m_cycleNodes.size() > 0);
1484         D_METHOD((
1485                 "NodeGroup::_cycleInit(%Ld)\n",
1486                 startTime));
1487
1488         // +++++ rescan latencies?
1489
1490         // figure new boundary & deadline from region length
1491         bigtime_t cyclePeriod = m_endPosition - m_startPosition;
1492
1493         if(cyclePeriod <= 0) {
1494                 // cycle region is no longer valid
1495                 m_cycleBoundary = 0LL;
1496                 m_cycleDeadline = 0LL;
1497
1498 //              no no no -- deadlocks when the thread calls this method
1499 //              // stop the thread
1500 //              _destroyCycleThread();
1501                 return;
1502         }
1503
1504         m_cycleStart = startTime;
1505         m_cycleBoundary = startTime + cyclePeriod;
1506         m_cycleDeadline = m_cycleBoundary - (m_cycleMaxLatency + s_rosterLatency);
1507 }
1508
1509
1510 // add a ref to the cycle set (in proper order, based on latency)
1511 void NodeGroup::_cycleAddRef(
1512         NodeRef*                                                                                ref) {
1513         assert_locked(this);
1514
1515         // make sure it's not already there
1516         ASSERT(find(
1517                 m_cycleNodes.begin(),
1518                 m_cycleNodes.end(),
1519                 ref) == m_cycleNodes.end());
1520
1521         // [re]calc latency if 0
1522         if(!ref->m_latency)
1523                 ref->_updateLatency();
1524
1525         node_set::iterator it;
1526         for(it = m_cycleNodes.begin();
1527                 it != m_cycleNodes.end(); ++it) {
1528                 if(ref->m_latency > (*it)->m_latency) {
1529                         m_cycleNodes.insert(it, ref);
1530                         break;
1531                 }
1532         }
1533
1534         // not inserted? new ref belongs at the end
1535         if(it == m_cycleNodes.end())
1536                 m_cycleNodes.insert(it, ref);
1537 }
1538
1539 // remove a ref from the cycle set
1540 void NodeGroup::_cycleRemoveRef(
1541         NodeRef*                                                                                ref) {
1542         assert_locked(this);
1543
1544         node_set::iterator it = find(
1545                 m_cycleNodes.begin(),
1546                 m_cycleNodes.end(),
1547                 ref);
1548         ASSERT(it != m_cycleNodes.end());
1549         m_cycleNodes.erase(it);
1550 }
1551
1552 bigtime_t NodeGroup::_cycleBoundary() const {
1553         Autolock _l(this);
1554         return m_cycleBoundary;
1555 }
1556
1557 // cycle thread impl.
1558 /*static*/
1559 status_t NodeGroup::_CycleThread(void* user) {
1560         ((NodeGroup*)user)->_cycleThread();
1561         return B_OK;
1562 }
1563
1564 void NodeGroup::_cycleThread() {
1565
1566         status_t err;
1567         int32 code;
1568         int32 errorCount = 0;
1569
1570         // +++++ liability -- if the thread has to be killed, this buffer
1571         //       won't be reclaimed
1572         char* msgBuffer = new char[_portMsgMaxSize];
1573         array_delete<char> _d(msgBuffer);
1574
1575         // create port
1576         ASSERT(!m_cyclePort);
1577         m_cyclePort = create_port(
1578                 _portLength,
1579                 _portName);
1580         ASSERT(m_cyclePort >= B_OK);
1581
1582         // the message-handling loop
1583         bool done = false;
1584         while(!done) {
1585
1586                 // *** wait until it's time to queue the next cycle, or until
1587                 // *** a message arrives
1588
1589                 lock();                         // **** BEGIN LOCKED SECTION ****
1590                 if(!_cycleValid()) {
1591                         unlock();
1592                         break;
1593                 }
1594
1595                 ASSERT(m_cycleNodes.size() > 0);
1596                 ASSERT(m_timeSourceObj);
1597
1598                 bigtime_t maxLatency = m_cycleNodes.front()->m_latency;
1599                 bigtime_t wakeUpAt = m_timeSourceObj->RealTimeFor(
1600                         m_cycleBoundary, maxLatency + s_rosterLatency);
1601                 bigtime_t timeout = wakeUpAt - m_timeSourceObj->RealTime();
1602
1603                 if(timeout <= 0) {
1604                         // +++++ whoops, I'm late.
1605                         // +++++ adjust to compensate !!!
1606                         PRINT((
1607                                 "*** NodeGroup::_cycleThread(): LATE\n"
1608                                 "    by %" B_PRIdBIGTIME "\n", -timeout));
1609                 }
1610
1611                 // +++++ if timeout is very short, spin until the target time arrives
1612
1613                 unlock();                       // **** END LOCKED SECTION ****
1614
1615                 // block until message arrives or it's time to wake up
1616                 err = read_port_etc(
1617                         m_cyclePort,
1618                         &code,
1619                         msgBuffer,
1620                         _portMsgMaxSize,
1621                         B_TIMEOUT,
1622                         timeout);
1623
1624                 if(err == B_TIMED_OUT) {
1625                         // the time has come to seek my nodes
1626                         _handleCycleService();
1627                         continue;
1628                 }
1629                 else if(err < B_OK) {
1630                         // any other error is bad news
1631                         PRINT((
1632                                 "* NodeGroup::_cycleThread(): read_port error:\n"
1633                                 "  %s\n"
1634                                 "  ABORTING\n\n", strerror(err)));
1635                         if(++errorCount > 10) {
1636                                 PRINT((
1637                                         "*** Too many errors; aborting.\n"));
1638                                 break;
1639                         }
1640                         continue;
1641                 }
1642
1643                 errorCount = 0;
1644
1645                 // process the message
1646                 switch(code) {
1647                         case _CYCLE_STOP:
1648                                 // bail
1649                                 done = true;
1650                                 break;
1651
1652                         case _CYCLE_END_CHANGED:
1653                         case _CYCLE_LATENCY_CHANGED:
1654                                 // fall through to next loop; for now, these messages
1655                                 // serve only to slap me out of my stupor and reassess
1656                                 // the timing situation...
1657                                 break;
1658
1659                         default:
1660                                 PRINT((
1661                                         "* NodeGroup::_cycleThread(): unknown message code '%"
1662                                                 B_PRId32 "'\n", code));
1663                                 break;
1664                 }
1665         } // while(!done)
1666
1667
1668         // delete port
1669         delete_port(m_cyclePort);
1670         m_cyclePort = 0;
1671
1672         // done
1673         m_cycleThreadDone = true;
1674 }
1675
1676 // cycle service: seek all nodes & initiate next cycle
1677 void NodeGroup::_handleCycleService() {
1678         Autolock _l(this);
1679 //      D_METHOD((
1680 //              "NodeGroup::_handleCycleService()\n"));
1681         status_t err;
1682
1683         if(!_cycleValid()) {
1684 //              PRINT((
1685 //                      "- _handleCycleService(): cycle not valid; quitting.\n"));
1686                 return;
1687         }
1688
1689         // seek
1690         for(node_set::iterator it = m_cycleNodes.begin();
1691                 it != m_cycleNodes.end(); ++it) {
1692                 err = (*it)->_seek(
1693                         m_startPosition,
1694                         m_cycleBoundary);
1695                 if(err < B_OK) {
1696                         PRINT((
1697                                 "- _handleCycleService(): node('%s')::_seek() failed:\n"
1698                                 "  %s\n",
1699                                 (*it)->name(), strerror(err)));
1700                 }
1701         }
1702
1703         // update cycle settings
1704         if(m_newStart) {
1705                 m_newStart = false;
1706                 m_startPosition = m_newStartPosition;
1707         }
1708         if(m_newEnd) {
1709                 m_newEnd = false;
1710                 m_endPosition = m_newEndPosition;
1711         }
1712
1713         // prepare next cycle
1714         _cycleInit(m_cycleBoundary);
1715 }
1716
1717 // END -- NodeGroup.cpp --