xref: /haiku/src/apps/cortex/NodeManager/NodeGroup.cpp (revision 68ea01249e1e2088933cb12f9c28d4e5c5d1c9ef)
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 --
1718