ports_test.cpp
Go to the documentation of this file.
1 /***************************************************************************
2  tag: Peter Soetens Mon Jun 26 13:26:02 CEST 2006 generictask_test.cpp
3 
4  generictask_test_3.cpp - description
5  -------------------
6  begin : Mon June 26 2006
7  copyright : (C) 2006 Peter Soetens
8  email : peter.soetens@fmtc.be
9 
10  ***************************************************************************
11  * *
12  * This program is free software; you can redistribute it and/or modify *
13  * it under the terms of the GNU General Public License as published by *
14  * the Free Software Foundation; either version 2 of the License, or *
15  * (at your option) any later version. *
16  * *
17  ***************************************************************************/
18 
19 #include "unit.hpp"
20 #include <InputPort.hpp>
21 #include <OutputPort.hpp>
22 
23 #include <TaskContext.hpp>
24 #include <extras/SlaveActivity.hpp>
25 #include <extras/SequentialActivity.hpp>
26 #include <extras/SimulationActivity.hpp>
27 #include <extras/SimulationThread.hpp>
28 
29 #include <boost/function_types/function_type.hpp>
30 #include <OperationCaller.hpp>
31 
32 #include <rtt-config.h>
33 
34 #include <memory>
35 
36 using namespace std;
37 using namespace RTT;
38 using namespace RTT::detail;
39 
40 
41 class EventPortsTC : public TaskContext
42 {
43 public:
44  bool had_event;
45  int nb_events;
46  EventPortsTC(): TaskContext("eptc") { resetStats(); }
47  void updateHook()
48  {
49  nb_events++;
50  had_event = true;
51  }
52  void resetStats() {
53  nb_events = 0;
54  had_event = false;
55  }
56 };
57 
61 class PortsTestFixture
62 {
63 public:
64  TaskContext* tc;
65  EventPortsTC* tce;
66  EventPortsTC* tc2;
67  EventPortsTC* tc3;
68  ActivityInterface* tsim;
69  ActivityInterface* stsim;
70  ActivityInterface* slsim;
71 
72  PortInterface* signalled_port;
73  void new_data_listener(PortInterface* port)
74  {
75  signalled_port = port;
76  }
77 
78 public:
79  PortsTestFixture()
80  {
81  tc = new TaskContext( "root", TaskContext::Stopped );
82  tce = new EventPortsTC();
83  tc2 = new EventPortsTC();
84  tc3 = new EventPortsTC();
85  tce->setActivity( new SequentialActivity() );
86  tc2->setActivity( new SequentialActivity() );
87  tc->setActivity( new SimulationActivity(0.001) );
88  tc3->setActivity( new SlaveActivity() );
89  tsim = tc->getActivity();
90  stsim = tc2->getActivity();
91  slsim = tc3->getActivity();
92  SimulationThread::Instance()->stop();
93  }
94 
95  ~PortsTestFixture()
96  {
97  // if ( tc->getPeer("programs") )
98  // delete tc->getPeer("programs");
99  tsim->stop();
100  stsim->stop();
101  delete tc;
102  delete tce;
103  delete tc2;
104  }
105 };
106 
107 
108 // Registers the fixture into the 'registry'
109 BOOST_FIXTURE_TEST_SUITE( PortsTestSuite, PortsTestFixture )
110 
111 BOOST_AUTO_TEST_CASE( testPortTaskInterface )
112 {
113  InputPort<int> rp1("Port1");
114  OutputPort<int> wp1("Port1");
115  InputPort<int> rp2("Port2");
116  OutputPort<int> wp2("Port2");
117 
118  tc->ports()->addPort( wp1 );
119  tc->ports()->addPort( rp2 );
120 
121  // check adding same port twice.
122  tc->ports()->addPort( wp1 );
123  {
124  // also check adding different port with same name.
125  InputPort<double> other_rp("Port1");
126  tc->ports()->addPort( other_rp );
127  // port will *not* autoremove itself... so do removePort or a crash will follow.
128  tc->ports()->removePort( other_rp.getName() );
129  }
130 
131  // We're adding the above ports to another TC as well.
132  // This is not supported behavior, as it will 'overtake' ownership,
133  {
134 #if __cplusplus > 199711L
135  unique_ptr<TaskContext> tc1(new TaskContext( "tc", TaskContext::Stopped ));
136  unique_ptr<TaskContext> tc2(new TaskContext( "tc2", TaskContext::Stopped ));
137 #else
138  auto_ptr<TaskContext> tc1(new TaskContext( "tc", TaskContext::Stopped ));
139  auto_ptr<TaskContext> tc2(new TaskContext( "tc2", TaskContext::Stopped ));
140 #endif
141 
142  tc1->ports()->addPort( rp1 );
143  tc1->ports()->addPort( wp2 );
144  tc2->ports()->addPort( rp2 );
145  tc2->ports()->addPort( wp1 );
146 
147  BOOST_CHECK( tc1->connectPorts(tc2.get()) );
148  BOOST_CHECK( wp1.connected() );
149  BOOST_CHECK( rp1.connected() );
150  BOOST_CHECK_EQUAL( wp1.write(2), WriteSuccess );
151  int value = 0;
152  BOOST_CHECK( rp1.read(value) );
153  BOOST_CHECK_EQUAL(2, value);
154 
155  BOOST_CHECK( wp2.connected() );
156  BOOST_CHECK( rp2.connected() );
157  BOOST_CHECK_EQUAL( wp2.write(3), WriteSuccess );
158  value = 0;
159  BOOST_CHECK( rp2.read(value) );
160  BOOST_CHECK_EQUAL(3, value);
161  }
162 
163  // Tasks have been destroyed, but the ports not. Automatic disconnection
164  // is done when port objects are disconnected
165  BOOST_CHECK( rp1.connected() );
166  BOOST_CHECK( rp2.connected() );
167  BOOST_CHECK( wp1.connected() );
168  BOOST_CHECK( wp2.connected() );
169 
170  // mandatory
171  tc->ports()->removePort( wp1.getName() ); // wp1 is not a port of tc, because it has been removed when another port with the same name was added
172  tc->ports()->removePort( rp2.getName() );
173 }
174 
175 BOOST_AUTO_TEST_CASE(testPortConnectionInitialization)
176 {
177  OutputPort<int> wp("WriterName", true);
178  InputPort<int> rp("ReaderName", ConnPolicy::data(ConnPolicy::LOCK_FREE, true));
179 
180  wp.setDataSample(-1);
181 
182  BOOST_CHECK( wp.createConnection(rp) );
183  int value = 0;
184  BOOST_CHECK( !rp.read(value) );
185  BOOST_CHECK_EQUAL( value, 0 );
186  BOOST_CHECK( !wp.getLastWrittenValue(value) );
187  rp.getDataSample(value);
188  BOOST_CHECK_EQUAL( -1, value );
189 
190  BOOST_CHECK_EQUAL( wp.write(10), WriteSuccess );
191  BOOST_CHECK( rp.read(value) );
192  BOOST_CHECK_EQUAL( 10, value );
193 
194  wp.disconnect(&rp);
195  BOOST_CHECK( !wp.connected() );
196  BOOST_CHECK( !rp.connected() );
197 
198  value = 0;
199  BOOST_CHECK( wp.getLastWrittenValue(value) );
200  BOOST_CHECK_EQUAL( 10, value );
201  BOOST_CHECK_EQUAL( 10, wp.getLastWrittenValue() );
202 
203  value = 0;
204  BOOST_CHECK( wp.createConnection(rp) );
205  BOOST_CHECK( rp.read(value) );
206  BOOST_CHECK_EQUAL( 10, value );
207  //wp.disconnect();
208 }
209 
210 BOOST_AUTO_TEST_CASE(testPortSimpleConnections)
211 {
212  OutputPort<int> wp("WriterName");
213  InputPort<int> rp("ReaderName", ConnPolicy::data(ConnPolicy::LOCK_FREE, false));
214 
215  BOOST_CHECK( !wp.connected() );
216  BOOST_CHECK( !rp.connected() );
217  {
218  int value;
219  BOOST_CHECK( !rp.read(value) );
220  BOOST_CHECK_EQUAL( wp.write(value), NotConnected ); // just checking if is works or if it crashes
221  }
222 
223  BOOST_REQUIRE( wp.createConnection(rp) );
224  BOOST_CHECK( wp.connected() );
225  BOOST_CHECK( rp.connected() );
226 
227  {
228  int value = 0;
229  BOOST_CHECK( !rp.read(value) );
230  BOOST_CHECK_EQUAL( wp.write(1), WriteSuccess );
231  BOOST_CHECK( rp.read(value) );
232  BOOST_CHECK( 1 == value );
233  }
234 
235  rp.clear();
236  {
237  int value = 0;
238  BOOST_CHECK( !rp.read(value) );
239  BOOST_CHECK_EQUAL( wp.write(1), WriteSuccess );
240  BOOST_CHECK( rp.read(value) );
241  BOOST_CHECK( 1 == value );
242  }
243 
244  // Try disconnecting starting at the writer. Disconnecting from the reader
245  // will be done later
246  wp.disconnect();
247  BOOST_CHECK( !wp.connected() );
248  BOOST_CHECK( !rp.connected() );
249  {
250  int value;
251  BOOST_CHECK_EQUAL( wp.write(value), NotConnected );
252  BOOST_CHECK( !rp.read(value) );
253  }
254  wp.disconnect(); // calling it when not connected should be fine as well
255 
256  {
257  int value = 0;
258  BOOST_CHECK( !rp.read(value) );
259  wp.createBufferConnection(rp, 4);
260  BOOST_CHECK( !rp.read(value) );
261  BOOST_CHECK_EQUAL( wp.write(1), WriteSuccess );
262  BOOST_CHECK_EQUAL( wp.write(2), WriteSuccess );
263  BOOST_CHECK_EQUAL( wp.write(3), WriteSuccess );
264  BOOST_CHECK_EQUAL( wp.write(4), WriteSuccess );
265  BOOST_CHECK_EQUAL( wp.write(5), WriteFailure ); // buffer full
266  BOOST_CHECK_EQUAL(0, value);
267  BOOST_CHECK( rp.read(value) );
268  BOOST_CHECK_EQUAL(1, value);
269  BOOST_CHECK( rp.read(value) );
270  BOOST_CHECK_EQUAL(2, value);
271 
272  rp.clear();
273  BOOST_CHECK_EQUAL( rp.read(value), NoData );
274  BOOST_CHECK_EQUAL( wp.write(10), WriteSuccess );
275  BOOST_CHECK_EQUAL( wp.write(20), WriteSuccess );
276  BOOST_CHECK( rp.read(value) );
277  BOOST_CHECK_EQUAL(10, value);
278  BOOST_CHECK( rp.read(value) );
279  BOOST_CHECK_EQUAL(20, value);
280  BOOST_CHECK_EQUAL( rp.read(value), OldData );
281  }
282 
283  // Try disconnecting from the reader this time
284  rp.disconnect();
285  BOOST_CHECK( !wp.connected() );
286  BOOST_CHECK( !rp.connected() );
287  {
288  int value;
289  BOOST_CHECK_EQUAL( wp.write(value), NotConnected );
290  BOOST_CHECK( !rp.read(value) );
291  }
292  rp.disconnect(); // calling it when not connected should be fine as well
293 
294  // Test automatic disconnection because of port destruction
295  {
296  InputPort<int> rp("ReaderName", ConnPolicy::data());
297  BOOST_CHECK(wp.createConnection(rp));
298  BOOST_CHECK( wp.connected() );
299  BOOST_CHECK( rp.connected() );
300  }
301  BOOST_CHECK( !wp.connected() );
302 }
303 
304 BOOST_AUTO_TEST_CASE(testPortOneWriterThreeReaders)
305 {
306  OutputPort<int> wp("W");
307  InputPort<int> rp1("R1", ConnPolicy::data());
308  InputPort<int> rp2("R2", ConnPolicy::buffer(4));
309  InputPort<int> rp3("R3", ConnPolicy::data());
310 
311  wp.createConnection(rp1);
312  BOOST_CHECK( wp.connected() );
313  BOOST_CHECK( rp1.connected() );
314  BOOST_CHECK( !rp2.connected() );
315  BOOST_CHECK( !rp3.connected() );
316  wp.createConnection(rp2);
317  BOOST_CHECK( wp.connected() );
318  BOOST_CHECK( rp1.connected() );
319  BOOST_CHECK( rp2.connected() );
320  BOOST_CHECK( !rp3.connected() );
321  wp.createConnection(rp3);
322  BOOST_CHECK( wp.connected() );
323  BOOST_CHECK( rp1.connected() );
324  BOOST_CHECK( rp2.connected() );
325  BOOST_CHECK( rp3.connected() );
326 
327  BOOST_CHECK_EQUAL( wp.write(10), WriteSuccess );
328  BOOST_CHECK_EQUAL( wp.write(15), WriteSuccess );
329  BOOST_CHECK_EQUAL( wp.write(20), WriteSuccess );
330  BOOST_CHECK_EQUAL( wp.write(25), WriteSuccess );
331  BOOST_CHECK_EQUAL( wp.write(30), WriteFailure ); // input buffer for R2 is full, but write to R1 and R3 was successful
332 
333  int value = 0;
334  BOOST_CHECK( rp1.read(value));
335  BOOST_CHECK_EQUAL(30, value);
336 
337  BOOST_CHECK( rp2.read(value));
338  BOOST_CHECK_EQUAL(10, value);
339  BOOST_CHECK( rp2.read(value));
340  BOOST_CHECK_EQUAL(15, value);
341  BOOST_CHECK( rp2.read(value));
342  BOOST_CHECK_EQUAL(20, value);
343  BOOST_CHECK( rp2.read(value));
344  BOOST_CHECK_EQUAL(25, value);
345  BOOST_CHECK_EQUAL( rp2.read(value), OldData);
346  BOOST_CHECK_EQUAL(25, value);
347 
348  BOOST_CHECK( rp3.read(value));
349  BOOST_CHECK_EQUAL(30, value);
350 
351  // Now removes only the buffer port
352  wp.disconnect(&rp2);
353  BOOST_CHECK_EQUAL( rp2.read(value), NoData );
354  BOOST_CHECK( wp.connected() );
355  BOOST_CHECK( rp1.connected() );
356  BOOST_CHECK( !rp2.connected() );
357  BOOST_CHECK( rp3.connected() );
358 
359  BOOST_CHECK_EQUAL( wp.write(10), WriteSuccess );
360  BOOST_CHECK( rp1.read(value));
361  BOOST_CHECK_EQUAL(10, value);
362  BOOST_CHECK( rp3.read(value));
363  BOOST_CHECK_EQUAL(10, value);
364 
365  // And finally the other ports as well
366  wp.disconnect(&rp1);
367  wp.disconnect(&rp3);
368  BOOST_CHECK( !wp.connected() );
369  BOOST_CHECK( !rp1.connected() );
370  BOOST_CHECK( !rp2.connected() );
371  BOOST_CHECK( !rp3.connected() );
372  BOOST_CHECK_EQUAL( rp1.read(value), NoData );
373  BOOST_CHECK_EQUAL( rp2.read(value), NoData );
374  BOOST_CHECK_EQUAL( rp3.read(value), NoData );
375 }
376 
377 BOOST_AUTO_TEST_CASE(testPortOneWriterThreeReadersWithSharedOutputBuffer)
378 {
379  ConnPolicy cp = ConnPolicy::buffer(4);
380  cp.buffer_policy = PerOutputPort;
381  OutputPort<int> wp("W");
382  InputPort<int> rp1("R1", cp);
383  InputPort<int> rp2("R2", cp);
384  InputPort<int> rp3("R3", cp);
385 
386  wp.createConnection(rp1);
387  BOOST_CHECK( wp.connected() );
388  BOOST_CHECK( rp1.connected() );
389  BOOST_CHECK( !rp2.connected() );
390  BOOST_CHECK( !rp3.connected() );
391  wp.createConnection(rp2);
392  BOOST_CHECK( wp.connected() );
393  BOOST_CHECK( rp1.connected() );
394  BOOST_CHECK( rp2.connected() );
395  BOOST_CHECK( !rp3.connected() );
396  wp.createConnection(rp3);
397  BOOST_CHECK( wp.connected() );
398  BOOST_CHECK( rp1.connected() );
399  BOOST_CHECK( rp2.connected() );
400  BOOST_CHECK( rp3.connected() );
401 
402  BOOST_CHECK_EQUAL( wp.write(10), WriteSuccess );
403  BOOST_CHECK_EQUAL( wp.write(15), WriteSuccess );
404  BOOST_CHECK_EQUAL( wp.write(20), WriteSuccess );
405  BOOST_CHECK_EQUAL( wp.write(25), WriteSuccess );
406  BOOST_CHECK_EQUAL( wp.write(30), WriteFailure ); // buffer full
407 
408  int value = 0;
409  BOOST_CHECK( rp1.read(value));
410  BOOST_CHECK_EQUAL(10, value);
411  BOOST_CHECK( rp2.read(value));
412  BOOST_CHECK_EQUAL(15, value);
413  BOOST_CHECK( rp3.read(value));
414  BOOST_CHECK_EQUAL(20, value);
415  BOOST_CHECK( rp3.read(value));
416  BOOST_CHECK_EQUAL(25, value);
417  // BOOST_CHECK_EQUAL(rp1.read(value), OldData); // WriteShared buffer connections never return OldData
418  BOOST_CHECK_EQUAL(rp1.read(value), NoData);
419  BOOST_CHECK_EQUAL(25, value);
420  // BOOST_CHECK_EQUAL(rp2.read(value), OldData); // WriteShared buffer connections never return OldData
421  BOOST_CHECK_EQUAL(rp2.read(value), NoData);
422  BOOST_CHECK_EQUAL(25, value);
423  // BOOST_CHECK_EQUAL(rp3.read(value), OldData); // WriteShared buffer connections never return OldData
424  BOOST_CHECK_EQUAL(rp3.read(value), NoData);
425  BOOST_CHECK_EQUAL(25, value);
426 
427  // Now removes only the R2
428  wp.disconnect(&rp2);
429  BOOST_CHECK_EQUAL( rp2.read(value), NoData );
430  BOOST_CHECK( wp.connected() );
431  BOOST_CHECK( rp1.connected() );
432  BOOST_CHECK( !rp2.connected() );
433  BOOST_CHECK( rp3.connected() );
434 
435  BOOST_CHECK_EQUAL( wp.write(10), WriteSuccess );
436  BOOST_CHECK( rp1.read(value) );
437  BOOST_CHECK_EQUAL(10, value);
438  // BOOST_CHECK_EQUAL( rp3.read(value), OldData ); // WriteShared buffer connections never return OldData
439  BOOST_CHECK_EQUAL( rp3.read(value), NoData );
440  BOOST_CHECK_EQUAL(10, value);
441 
442  // And finally the other ports as well
443  wp.disconnect(&rp1);
444  wp.disconnect(&rp3);
445  BOOST_CHECK( !wp.connected() );
446  BOOST_CHECK( !rp1.connected() );
447  BOOST_CHECK( !rp2.connected() );
448  BOOST_CHECK( !rp3.connected() );
449  BOOST_CHECK_EQUAL( rp1.read(value), NoData );
450  BOOST_CHECK_EQUAL( rp2.read(value), NoData );
451  BOOST_CHECK_EQUAL( rp3.read(value), NoData );
452 }
453 
454 BOOST_AUTO_TEST_CASE(testPortThreeWritersOneReaderWithSharedInputBuffer)
455 {
456  ConnPolicy cp = ConnPolicy::buffer(4);
457  cp.buffer_policy = PerInputPort;
458  OutputPort<int> wp1("W1");
459  OutputPort<int> wp2("W2");
460  OutputPort<int> wp3("W3");
461  InputPort<int> rp("R", cp);
462 
463  wp1.createConnection(rp);
464  BOOST_CHECK( wp1.connected() );
465  BOOST_CHECK( rp.connected() );
466  BOOST_CHECK( !wp2.connected() );
467  BOOST_CHECK( !wp3.connected() );
468  wp2.createConnection(rp);
469  BOOST_CHECK( rp.connected() );
470  BOOST_CHECK( wp1.connected() );
471  BOOST_CHECK( wp2.connected() );
472  BOOST_CHECK( !wp3.connected() );
473  wp3.createConnection(rp);
474  BOOST_CHECK( rp.connected() );
475  BOOST_CHECK( wp1.connected() );
476  BOOST_CHECK( wp2.connected() );
477  BOOST_CHECK( wp3.connected() );
478 
479  BOOST_CHECK( rp.getSharedBuffer() );
480 
481  BOOST_CHECK_EQUAL( wp1.write(10), WriteSuccess );
482  BOOST_CHECK_EQUAL( wp1.write(20), WriteSuccess );
483  BOOST_CHECK_EQUAL( wp1.write(30), WriteSuccess );
484  BOOST_CHECK_EQUAL( wp1.write(40), WriteSuccess );
485  BOOST_CHECK_EQUAL( wp1.write(50), WriteFailure ); // buffer full
486 
487  BOOST_CHECK_EQUAL( wp2.write(12), WriteFailure ); // buffer full
488  BOOST_CHECK_EQUAL( wp2.write(22), WriteFailure ); // buffer full
489  BOOST_CHECK_EQUAL( wp2.write(32), WriteFailure ); // buffer full
490  BOOST_CHECK_EQUAL( wp2.write(42), WriteFailure ); // buffer full
491  BOOST_CHECK_EQUAL( wp2.write(52), WriteFailure ); // buffer full
492 
493  BOOST_CHECK_EQUAL( wp3.write(13), WriteFailure ); // buffer full
494  BOOST_CHECK_EQUAL( wp3.write(23), WriteFailure ); // buffer full
495  BOOST_CHECK_EQUAL( wp3.write(33), WriteFailure ); // buffer full
496  BOOST_CHECK_EQUAL( wp3.write(43), WriteFailure ); // buffer full
497  BOOST_CHECK_EQUAL( wp3.write(53), WriteFailure ); // buffer full
498 
499  int value = 0;
500  BOOST_CHECK( rp.read(value));
501  BOOST_CHECK_EQUAL(10, value);
502  BOOST_CHECK( rp.read(value));
503  BOOST_CHECK_EQUAL(20, value);
504  BOOST_CHECK( rp.read(value));
505  BOOST_CHECK_EQUAL(30, value);
506  BOOST_CHECK( rp.read(value));
507  BOOST_CHECK_EQUAL(40, value);
508  // rp's input buffer is empty now.
509 
510  // Now removes the middle writer
511  wp2.disconnect(&rp);
512  BOOST_CHECK( rp.connected() );
513  BOOST_CHECK( wp1.connected() );
514  BOOST_CHECK( !wp2.connected() );
515  BOOST_CHECK( wp3.connected() );
516 
517  // write one more sample
518  BOOST_CHECK_EQUAL( wp1.write(60), WriteSuccess );
519  BOOST_CHECK_EQUAL( rp.read(value), NewData );
520  BOOST_CHECK_EQUAL(60, value);
521  // rp's input buffer is empty now.
522 
523  // now check that the data written to wp3 was indeed dropped:
524  BOOST_CHECK_EQUAL( rp.read(value), OldData );
525  BOOST_CHECK_EQUAL(60, value);
526 
527  // And finally the other ports as well
528  rp.disconnect(&wp1);
529  rp.disconnect(&wp3);
530  BOOST_CHECK( !rp.connected() );
531  BOOST_CHECK( !wp1.connected() );
532  BOOST_CHECK( !wp2.connected() );
533  BOOST_CHECK( !wp3.connected() );
534  BOOST_CHECK( !rp.getSharedBuffer() );
535  BOOST_CHECK_EQUAL( rp.read(value), NoData );
536 }
537 
538 BOOST_AUTO_TEST_CASE(testPortThreeWritersOneReader)
539 {
540  OutputPort<int> wp1("W1");
541  OutputPort<int> wp2("W2");
542  OutputPort<int> wp3("W3");
543  InputPort<int> rp("R", ConnPolicy::buffer(4));
544 
545  wp1.createConnection(rp);
546  BOOST_CHECK( wp1.connected() );
547  BOOST_CHECK( rp.connected() );
548  BOOST_CHECK( !wp2.connected() );
549  BOOST_CHECK( !wp3.connected() );
550  wp2.createConnection(rp);
551  BOOST_CHECK( rp.connected() );
552  BOOST_CHECK( wp1.connected() );
553  BOOST_CHECK( wp2.connected() );
554  BOOST_CHECK( !wp3.connected() );
555  wp3.createConnection(rp);
556  BOOST_CHECK( rp.connected() );
557  BOOST_CHECK( wp1.connected() );
558  BOOST_CHECK( wp2.connected() );
559  BOOST_CHECK( wp3.connected() );
560  BOOST_CHECK( !rp.getSharedBuffer() );
561 
562  BOOST_CHECK_EQUAL( wp1.write(10), WriteSuccess );
563  BOOST_CHECK_EQUAL( wp1.write(20), WriteSuccess );
564  BOOST_CHECK_EQUAL( wp1.write(30), WriteSuccess );
565  BOOST_CHECK_EQUAL( wp1.write(40), WriteSuccess );
566  BOOST_CHECK_EQUAL( wp1.write(50), WriteFailure ); // buffer full
567 
568  BOOST_CHECK_EQUAL( wp2.write(12), WriteSuccess );
569  BOOST_CHECK_EQUAL( wp2.write(22), WriteSuccess );
570  BOOST_CHECK_EQUAL( wp2.write(32), WriteSuccess );
571  BOOST_CHECK_EQUAL( wp2.write(42), WriteSuccess );
572  BOOST_CHECK_EQUAL( wp2.write(52), WriteFailure ); // buffer full
573 
574  BOOST_CHECK_EQUAL( wp3.write(13), WriteSuccess );
575  BOOST_CHECK_EQUAL( wp3.write(23), WriteSuccess );
576  BOOST_CHECK_EQUAL( wp3.write(33), WriteSuccess );
577  BOOST_CHECK_EQUAL( wp3.write(43), WriteSuccess );
578  BOOST_CHECK_EQUAL( wp3.write(53), WriteFailure ); // buffer full
579 
580  int value = 0;
581  BOOST_CHECK( rp.read(value));
582  BOOST_CHECK_EQUAL(10, value);
583  BOOST_CHECK( rp.read(value));
584  BOOST_CHECK_EQUAL(20, value);
585  BOOST_CHECK( rp.read(value));
586  BOOST_CHECK_EQUAL(30, value);
587  BOOST_CHECK( rp.read(value));
588  BOOST_CHECK_EQUAL(40, value);
589 
590  // Now removes the middle writer
591  wp2.disconnect(&rp);
592  BOOST_CHECK( rp.connected() );
593  BOOST_CHECK( wp1.connected() );
594  BOOST_CHECK( !wp2.connected() );
595  BOOST_CHECK( wp3.connected() );
596 
597  // write one more sample
598  BOOST_CHECK_EQUAL( wp1.write(60), WriteSuccess );
599  BOOST_CHECK( rp.read(value));
600  BOOST_CHECK_EQUAL(60, value);
601 
602  // now check if wp3's connection is used:
603  BOOST_CHECK( rp.read(value));
604  BOOST_CHECK_EQUAL(13, value);
605  BOOST_CHECK( rp.read(value));
606  BOOST_CHECK_EQUAL(23, value);
607 
608  // in the middle adding a sample
609  BOOST_CHECK_EQUAL( wp1.write(70), WriteSuccess );
610 
611  BOOST_CHECK( rp.read(value));
612  BOOST_CHECK_EQUAL(33, value);
613  BOOST_CHECK( rp.read(value));
614  BOOST_CHECK_EQUAL(43, value);
615 
616  // now the in the middle sample shows up
617  BOOST_CHECK_EQUAL( rp.read(value), NewData );
618  BOOST_CHECK_EQUAL(70, value);
619  value = 0;
620  BOOST_CHECK_EQUAL( rp.read(value), OldData );
621  BOOST_CHECK_EQUAL(70, value);
622 
623  // And finally the other ports as well
624  rp.disconnect(&wp1);
625  rp.disconnect(&wp3);
626  BOOST_CHECK( !rp.connected() );
627  BOOST_CHECK( !wp1.connected() );
628  BOOST_CHECK( !wp2.connected() );
629  BOOST_CHECK( !wp3.connected() );
630  BOOST_CHECK_EQUAL( rp.read(value), NoData );
631 }
632 
633 BOOST_AUTO_TEST_CASE(testSharedBufferConnection)
634 {
635  ConnPolicy cp = ConnPolicy::buffer(10);
636  cp.buffer_policy = Shared;
637 
638  OutputPort<int> wp1("W1");
639  OutputPort<int> wp2("W1");
640  InputPort<int> rp1("R1", cp);
641  InputPort<int> rp2("R2", cp);
642  int value = 0;
643 
644  BOOST_CHECK( wp1.createConnection(rp1) );
645  BOOST_CHECK( wp1.createConnection(rp2) );
646  BOOST_CHECK( wp2.createConnection(rp1) );
647  BOOST_CHECK( wp2.createConnection(rp2) );
648 
649  BOOST_CHECK_EQUAL( wp1.write(11), WriteSuccess );;
650  BOOST_CHECK_EQUAL( rp1.read(value), NewData );
651  BOOST_CHECK_EQUAL(11, value);
652  BOOST_CHECK_EQUAL( rp2.read(value), NoData );
653  BOOST_CHECK_EQUAL(11, value);
654 
655  BOOST_CHECK_EQUAL( wp1.write(12), WriteSuccess );;
656  BOOST_CHECK_EQUAL( rp2.read(value), NewData );
657  BOOST_CHECK_EQUAL(12, value);
658  BOOST_CHECK_EQUAL( rp1.read(value), NoData );
659  BOOST_CHECK_EQUAL(12, value);
660 
661  BOOST_CHECK_EQUAL( wp2.write(21), WriteSuccess );;
662  BOOST_CHECK_EQUAL( rp1.read(value), NewData );
663  BOOST_CHECK_EQUAL(21, value);
664  BOOST_CHECK_EQUAL( rp2.read(value), NoData );
665  BOOST_CHECK_EQUAL(21, value);
666 
667  BOOST_CHECK_EQUAL( wp2.write(22), WriteSuccess );;
668  BOOST_CHECK_EQUAL( rp2.read(value), NewData );
669  BOOST_CHECK_EQUAL(22, value);
670  BOOST_CHECK_EQUAL( rp1.read(value), NoData );
671  BOOST_CHECK_EQUAL(22, value);
672 
673  BOOST_CHECK_EQUAL( wp1.write(31), WriteSuccess );;
674  BOOST_CHECK_EQUAL( wp2.write(32), WriteSuccess );;
675  BOOST_CHECK_EQUAL( wp1.write(33), WriteSuccess );;
676  BOOST_CHECK_EQUAL( wp2.write(34), WriteSuccess );;
677  BOOST_CHECK_EQUAL( rp1.read(value), NewData );
678  BOOST_CHECK_EQUAL(31, value);
679  BOOST_CHECK_EQUAL( rp2.read(value), NewData );
680  BOOST_CHECK_EQUAL(32, value);
681  BOOST_CHECK_EQUAL( rp2.read(value), NewData );
682  BOOST_CHECK_EQUAL(33, value);
683  BOOST_CHECK_EQUAL( rp1.read(value), NewData );
684  BOOST_CHECK_EQUAL(34, value);
685  BOOST_CHECK_EQUAL( rp1.read(value), NoData );
686  BOOST_CHECK_EQUAL(34, value);
687  BOOST_CHECK_EQUAL( rp2.read(value), NoData );
688  BOOST_CHECK_EQUAL(34, value);
689 }
690 
691 BOOST_AUTO_TEST_CASE(testSharedDataConnection)
692 {
693  ConnPolicy cp = ConnPolicy::data(ConnPolicy::LOCKED);
694  cp.buffer_policy = Shared;
695 
696  OutputPort<int> wp1("W1");
697  OutputPort<int> wp2("W1");
698  InputPort<int> rp1("R1", cp);
699  InputPort<int> rp2("R2", cp);
700  int value = 0;
701 
702  BOOST_CHECK( wp1.createConnection(rp1) );
703  BOOST_CHECK( wp1.createConnection(rp2) );
704  BOOST_CHECK( wp2.createConnection(rp1) );
705  BOOST_CHECK( wp2.createConnection(rp2) );
706 
707  // same as in testSharedBufferConnection, but different expectations
708  BOOST_CHECK_EQUAL( wp1.write(31), WriteSuccess );
709  BOOST_CHECK_EQUAL( wp2.write(32), WriteSuccess );
710  BOOST_CHECK_EQUAL( wp1.write(33), WriteSuccess );
711  BOOST_CHECK_EQUAL( wp2.write(34), WriteSuccess );
712  BOOST_CHECK_EQUAL( rp1.read(value), NewData );
713  BOOST_CHECK_EQUAL(34, value);
714  BOOST_CHECK_EQUAL( rp2.read(value), OldData );
715  BOOST_CHECK_EQUAL(34, value);
716  BOOST_CHECK_EQUAL( rp2.read(value), OldData );
717  BOOST_CHECK_EQUAL(34, value);
718  BOOST_CHECK_EQUAL( rp1.read(value), OldData );
719  BOOST_CHECK_EQUAL(34, value);
720  BOOST_CHECK_EQUAL( rp1.read(value), OldData );
721  BOOST_CHECK_EQUAL(34, value);
722  BOOST_CHECK_EQUAL( rp2.read(value), OldData );
723  BOOST_CHECK_EQUAL(34, value);
724 }
725 
726 
727 BOOST_AUTO_TEST_CASE(testInvalidReadPolicyConnections)
728 {
729  OutputPort<int> wp1("W1");
730  OutputPort<int> wp2("W2");
731  InputPort<int> rp("R1");
732 
733  // mix buffer policies
734  ConnPolicy cp_PerConnection, cp_PerInputPort;
735  cp_PerConnection.buffer_policy = PerConnection;
736  cp_PerInputPort.buffer_policy = PerInputPort;
737  BOOST_CHECK( wp1.connectTo(&rp, cp_PerConnection) );
738  BOOST_CHECK( !wp2.connectTo(&rp, cp_PerInputPort) );
739  rp.disconnect();
740 
741  // mix data and buffer connections with PerInputPort read policy
742  ConnPolicy cp_DATA = ConnPolicy::data(ConnPolicy::LOCKED);
743  ConnPolicy cp_BUFFER = ConnPolicy::buffer(10, ConnPolicy::LOCKED);
744  cp_DATA.buffer_policy = PerInputPort;
745  cp_BUFFER.buffer_policy = PerInputPort;
746  BOOST_CHECK( wp1.connectTo(&rp, cp_DATA) );
747  BOOST_CHECK( !wp2.connectTo(&rp, cp_BUFFER) );
748  rp.disconnect();
749 
750  // mix different locking policies with PerInputPort read policy
751  ConnPolicy cp_LOCK_FREE = ConnPolicy::buffer(10, ConnPolicy::LOCK_FREE);
752  ConnPolicy cp_LOCKED = ConnPolicy::buffer(10, ConnPolicy::LOCKED);
753  cp_LOCK_FREE.buffer_policy = PerInputPort;
754  cp_LOCKED.buffer_policy = PerInputPort;
755  BOOST_CHECK( wp1.connectTo(&rp, cp_LOCK_FREE) );
756  BOOST_CHECK( !wp2.connectTo(&rp, cp_LOCKED) );
757  rp.disconnect();
758 
759  // mix different buffer sizes with PerInputPort read policy
760  ConnPolicy cp_BUFFER5 = ConnPolicy::buffer(5);
761  ConnPolicy cp_BUFFER10 = ConnPolicy::buffer(10);
762  cp_BUFFER5.buffer_policy = PerInputPort;
763  cp_BUFFER10.buffer_policy = PerInputPort;
764  BOOST_CHECK( wp1.connectTo(&rp, cp_BUFFER5) );
765  BOOST_CHECK( !wp2.connectTo(&rp, cp_BUFFER10) );
766  rp.disconnect();
767 }
768 
769 BOOST_AUTO_TEST_CASE(testInvalidSharedConnection)
770 {
771  ConnPolicy cp = ConnPolicy::data(ConnPolicy::LOCKED);
772  cp.buffer_policy = Shared;
773 
774  OutputPort<int> wp1("W1");
775  OutputPort<int> wp2("W1");
776  InputPort<int> rp1("R1", cp);
777  InputPort<int> rp2("R2", cp);
778 
779  BOOST_CHECK( wp1.createConnection(rp1) ); // new shared connection
780  BOOST_CHECK( wp2.createConnection(rp2) ); // new shared connection
781  BOOST_CHECK( !wp1.createConnection(rp2) ); // different connection => failure
782  BOOST_CHECK( !wp2.createConnection(rp1) ); // different connection => failure
783 }
784 
785 BOOST_AUTO_TEST_CASE( testPortObjects)
786 {
787  OutputPort<double> wp1("Write");
788  InputPort<double> rp1("Read");
789 
790  tc->ports()->addPort( wp1 );
791  tc->ports()->addPort( rp1 );
792 
793  // Check if ports were added as objects as well
794  BOOST_CHECK( tc->provides("Write") != 0 );
795  BOOST_CHECK( tc->provides("Read") != 0 );
796 
797  // Set initial value
798  BOOST_CHECK_EQUAL( wp1.write( 1.0 ), NotConnected );
799 
800  // Connect ports.
801  wp1.createConnection( rp1 );
802 
803  // Test OperationCallers set/get
804  OperationCaller<WriteStatus(double const&)> mset;
805  OperationCaller<FlowStatus(double&)> mget;
806 
807  mset = tc->provides("Write")->getOperation("write");
808  BOOST_CHECK( mset.ready() );
809 
810  mget = tc->provides("Read")->getOperation("read");
811  BOOST_CHECK( mget.ready() );
812 
813  BOOST_CHECK_EQUAL( mset( 3.991 ), WriteSuccess );
814 
815  double get_value = 0;
816  BOOST_CHECK( mget(get_value) );
817  BOOST_CHECK_CLOSE( 3.991, get_value, 0.001 );
818 
820  tc->ports()->removePort("Read");
821  BOOST_CHECK( tc->provides()->hasService("Read") == 0 );
822  BOOST_CHECK( tc->ports()->getPort("Read") == 0 );
823 
824  tc->ports()->removePort("Write");
825  BOOST_CHECK( tc->provides()->hasService("Write") == 0 );
826  BOOST_CHECK( tc->ports()->getPort("Write") == 0 );
827 }
828 
829 #ifdef ORO_SIGNALLING_PORTS
830 BOOST_AUTO_TEST_CASE(testPortSignalling)
831 {
832  OutputPort<double> wp1("Write");
833  InputPort<double> rp1("Read");
834 
835  Handle hl( rp1.getNewDataOnPortEvent()->setup(
836  boost::bind(&PortsTestFixture::new_data_listener, this, _1) ) );
837  hl.connect();
838 
839  wp1.createConnection(rp1, ConnPolicy::data());
840  signalled_port = 0;
841  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
842  BOOST_CHECK(&rp1 == signalled_port);
843 
844  wp1.disconnect();
845  wp1.createConnection(rp1, ConnPolicy::buffer(2));
846  signalled_port = 0;
847  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
848  BOOST_CHECK(&rp1 == signalled_port);
849  signalled_port = 0;
850  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
851  BOOST_CHECK(&rp1 == signalled_port);
852  signalled_port = 0;
853  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
854  BOOST_CHECK(0 == signalled_port);
855 }
856 #endif
857 
858 BOOST_AUTO_TEST_CASE(testPortAddRemove)
859 {
860  OutputPort<double>* wp1 = new OutputPort<double>("Write");
861  InputPort<double>* rp1 = new InputPort<double>("Read");
862  InputPort<double>* ep1 = new InputPort<double>("ERead");
863  TaskContext tc("tc");
864  tc.addPort( *wp1 );
865  tc.addEventPort( *rp1, boost::bind(&PortsTestFixture::new_data_listener, this, _1) );
866  tc.addEventPort( *ep1 );
867 
868  wp1->createConnection(*rp1, ConnPolicy::data());
869  wp1->createConnection(*ep1, ConnPolicy::data());
870 
871  tc.start();
872  wp1->write(0.1);
873  tc.stop();
874 
875  tc.ports()->removePort("Write");
876  tc.ports()->removePort("Read");
877  tc.ports()->removePort("ERead");
878  BOOST_CHECK(wp1->getInterface() == 0);
879  BOOST_CHECK(rp1->getInterface() == 0);
880  BOOST_CHECK(ep1->getInterface() == 0);
881 
882  wp1->write(0.1);
883 
884  delete wp1;
885  delete rp1;
886  delete ep1;
887 
888  tc.start();
889  tc.stop();
890 
891  wp1 = new OutputPort<double>("Write");
892  rp1 = new InputPort<double>("Read");
893  ep1 = new InputPort<double>("ERead");
894 
895  tc.addPort( *wp1 );
896  tc.addEventPort( *rp1, boost::bind(&PortsTestFixture::new_data_listener, this, _1) );
897  tc.addEventPort( *ep1 );
898 
899  wp1->createConnection(*rp1, ConnPolicy::data());
900  wp1->createConnection(*ep1, ConnPolicy::data());
901 
902  tc.start();
903  wp1->write(0.1);
904  tc.stop();
905 
906  tc.ports()->removePort("Write");
907  tc.ports()->removePort("Read");
908  tc.ports()->removePort("ERead");
909 
910  wp1->write(0.1);
911 
912  delete wp1;
913  delete rp1;
914  delete ep1;
915 
916  tc.start();
917  tc.stop();
918 }
919 
920 BOOST_AUTO_TEST_CASE(testEventPortSignalling)
921 {
922  OutputPort<double> wp1("Write");
923  InputPort<double> rp1("Read");
924 
925  BOOST_REQUIRE(tce->configure());
926  BOOST_REQUIRE(tce->isConfigured());
927 
928  BOOST_REQUIRE(slsim->isActive());
929 
930 
931  tce->start();
932  tce->resetStats();
933 
934  tce->addEventPort(rp1,boost::bind(&PortsTestFixture::new_data_listener, this, _1) );
935 
936  BOOST_CHECK( slsim->execute() );
937 
938  wp1.createConnection(rp1, ConnPolicy::data());
939  signalled_port = 0;
940  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
941  BOOST_CHECK(&rp1 == signalled_port);
942  BOOST_CHECK(tce->had_event);
943  tce->resetStats();
944 
945  wp1.disconnect();
946  wp1.createConnection(rp1, ConnPolicy::buffer(2));
947  // send two items into the buffer
948  signalled_port = 0;
949  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
950  BOOST_CHECK(&rp1 == signalled_port);
951  BOOST_CHECK(tce->had_event);
952  tce->resetStats();
953  signalled_port = 0;
954  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
955  BOOST_CHECK(&rp1 == signalled_port);
956  BOOST_CHECK(tce->had_event);
957  tce->resetStats();
958  signalled_port = 0;
959  // test buffer full:
960  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteFailure );
961 
962  BOOST_CHECK( slsim->execute() );
963  BOOST_CHECK(0 == signalled_port);
964  BOOST_CHECK( !tce->had_event);
965  tce->resetStats();
966 
967  // mandatory
968  tce->ports()->removePort( rp1.getName() );
969 }
970 
971 
972 BOOST_AUTO_TEST_CASE(testEventPortSignallingFromSlave)
973 {
974  OutputPort<double> wp1("Write");
975  InputPort<double> rp1("Read");
976 
977  tc3->start();
978  tc3->resetStats();
979 
980  tc3->addEventPort(rp1,boost::bind(&PortsTestFixture::new_data_listener, this, _1) );
981 
982  wp1.createConnection(rp1, ConnPolicy::data());
983  signalled_port = 0;
984  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
985 
986  BOOST_CHECK( slsim->execute() );
987  BOOST_CHECK(&rp1 == signalled_port);
988  BOOST_CHECK(tc3->had_event);
989  tc3->resetStats();
990 
991  wp1.disconnect();
992  wp1.createConnection(rp1, ConnPolicy::buffer(2));
993  // send two items into the buffer
994  signalled_port = 0;
995  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
996 
997  BOOST_CHECK( slsim->execute() );
998  BOOST_CHECK(&rp1 == signalled_port);
999  BOOST_CHECK(tc3->had_event);
1000  tc3->resetStats();
1001  signalled_port = 0;
1002  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
1003 
1004  BOOST_CHECK( slsim->execute() );
1005  BOOST_CHECK(&rp1 == signalled_port);
1006  BOOST_CHECK(tc3->had_event);
1007  tc3->resetStats();
1008  signalled_port = 0;
1009  // test buffer full (updateHook called due to execute, but no callback executed):
1010  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteFailure );
1011  BOOST_CHECK( slsim->execute() );
1012  BOOST_CHECK(0 == signalled_port);
1013  BOOST_CHECK( tc3->had_event);
1014  // empty one element and try again:
1015  double d;
1016  rp1.read(d);
1017  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
1018  BOOST_CHECK( slsim->execute() );
1019  BOOST_CHECK(&rp1 == signalled_port);
1020  tc3->resetStats();
1021 
1022  // mandatory
1023  tc3->ports()->removePort( rp1.getName() );
1024 }
1025 
1026 BOOST_AUTO_TEST_CASE(testPlainPortNotSignalling)
1027 {
1028  OutputPort<double> wp1("Write");
1029  InputPort<double> rp1("Read");
1030 
1031  tce->start();
1032  tce->resetStats();
1033 
1034  tce->addPort(rp1);
1035 
1036  wp1.createConnection(rp1, ConnPolicy::data());
1037  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
1038  BOOST_CHECK( !tce->had_event );
1039  tce->resetStats();
1040 
1041  wp1.disconnect();
1042  wp1.createConnection(rp1, ConnPolicy::buffer(2));
1043  BOOST_CHECK_EQUAL( wp1.write(0.1), WriteSuccess );
1044  BOOST_CHECK( !tce->had_event );
1045 
1046  // mandatory
1047  tce->ports()->removePort( rp1.getName() );
1048 }
1049 
1050 BOOST_AUTO_TEST_CASE(testPortDataSource)
1051 {
1052  OutputPort<int> wp1("Write");
1053 #if __cplusplus > 199711L
1054  unique_ptr<InputPortInterface>
1055 #else
1056  auto_ptr<InputPortInterface>
1057 #endif
1058  reader(dynamic_cast<InputPortInterface*>(wp1.antiClone()));
1059  BOOST_CHECK(wp1.connectTo(&*reader, ConnPolicy::buffer(2)));
1060 
1061  DataSource<int>::shared_ptr source = static_cast< DataSource<int>* >(reader->getDataSource());
1062  BOOST_CHECK(source);
1063 
1064  BOOST_CHECK(!source->evaluate());
1065  BOOST_CHECK_EQUAL( wp1.write(10), WriteSuccess );
1066  BOOST_CHECK_EQUAL( wp1.write(20), WriteSuccess );
1067  // value is still null when not get()/evaluate()
1068  BOOST_CHECK_EQUAL(0, source->value());
1069 
1070  // read a sample:
1071  BOOST_CHECK(source->evaluate());
1072  BOOST_CHECK_EQUAL(10, source->value());
1073  BOOST_CHECK_EQUAL(10, source->value());
1074 
1075  // get a sample (=evaluate+value):
1076  BOOST_CHECK_EQUAL(20, source->get());
1077 
1078  // buffer empty, but value remains same as old:
1079  BOOST_CHECK(!source->evaluate());
1080  BOOST_CHECK_EQUAL(0, source->get());
1081  BOOST_CHECK_EQUAL(20, source->value());
1082 }
1083 
1084 BOOST_AUTO_TEST_SUITE_END()
1085 
1086 
1089 class ConcurrencyPortsTestFixture
1090 {
1091 public:
1092  template <typename T>
1093  class PortWriterThread : public Activity
1094  {
1095  public:
1096  PortWriterThread(const T& sample = T())
1097  : port("writer"), sample(sample), step_counter(0), write_counter(0)
1098  {}
1099  void step() {
1100  if (port.write(sample) == WriteSuccess) {
1101  ++write_counter;
1102  }
1103  this->trigger();
1104  this->yield();
1105  ++step_counter;
1106  }
1107  bool breakLoop() { return true; }
1108  public:
1109  OutputPort<T> port;
1110  T sample;
1111  unsigned step_counter;
1112  unsigned write_counter;
1113  };
1114 
1115  template <typename T>
1116  class PortReaderThread : public Activity
1117  {
1118  public:
1119  PortReaderThread()
1120  : port("reader"), step_counter(0), read_counter(0)
1121  {}
1122  void step() {
1123  if (port.read(sample) == NewData) {
1124  ++read_counter;
1125  }
1126  this->trigger();
1127  this->yield();
1128  ++step_counter;
1129  }
1130  bool breakLoop() { return true; }
1131  public:
1132  InputPort<T> port;
1133  T sample;
1134  unsigned step_counter;
1135  unsigned read_counter;
1136  };
1137 
1138  class PortConnectorThread : public Activity
1139  {
1140  public:
1141  PortConnectorThread(PortInterface &connect_port, PortInterface &other_port, const ConnPolicy &policy = ConnPolicy())
1142  : connect_port(connect_port), other_port(other_port), policy(policy),
1143  step_counter(0), connect_counter(0), connect_failure_counter(0),
1144  disconnect_counter(0), disconnect_failure_counter(0), connected(false)
1145  {}
1146  void step() {
1147  if (!connected) {
1148  if ((!shared_connection && connect_port.connectTo(&other_port, policy)) ||
1149  (shared_connection && connect_port.createConnection(shared_connection, policy))) {
1150  connected = true;
1151  ++connect_counter;
1152  } else {
1153  ++connect_failure_counter;
1154  }
1155  } else {
1156  if ((!shared_connection && connect_port.disconnect(&other_port)) ||
1157  (shared_connection && connect_port.getManager()->removeConnection(shared_connection.get()))) {
1158  connected = false;
1159  ++disconnect_counter;
1160  } else {
1161  ++disconnect_failure_counter;
1162  }
1163  }
1164  this->trigger();
1165  this->yield();
1166  ++step_counter;
1167  }
1168  bool breakLoop() { return true; }
1169  public:
1170  PortInterface &connect_port;
1171  PortInterface &other_port;
1172  RTT::internal::SharedConnectionBase::shared_ptr shared_connection;
1173  ConnPolicy policy;
1174  unsigned step_counter;
1175  unsigned connect_counter, connect_failure_counter;
1176  unsigned disconnect_counter, disconnect_failure_counter;
1177  bool connected;
1178  };
1179 
1180  typedef double T;
1181  PortWriterThread<T> writer;
1182  PortReaderThread<T> reader;
1183  PortConnectorThread connector;
1184  OutputPort<T> another_output_port;
1185  InputPort<T> another_input_port;
1186  PortConnectorThread another_output_connector, another_input_connector;
1187 
1188 public:
1189  ConcurrencyPortsTestFixture()
1190  : writer()
1191  , reader()
1192  , connector(writer.port, reader.port)
1193  , another_output_port("another_output_port")
1194  , another_input_port("another_input_port")
1195  , another_output_connector(another_output_port, reader.port)
1196  , another_input_connector(another_input_port, writer.port)
1197  {}
1198 
1199  ~ConcurrencyPortsTestFixture()
1200  {
1201  stop();
1202  disconnect();
1203  }
1204 
1205  bool connect(const ConnPolicy &policy) {
1206  connector.policy = policy;
1207  another_output_connector.policy = policy;
1208  another_input_connector.policy = policy;
1209  return writer.port.connectTo(&reader.port, policy);
1210  }
1211 
1212  void disconnect() {
1213  writer.port.disconnect();
1214  reader.port.disconnect();
1215  another_output_port.disconnect();
1216  another_input_port.disconnect();
1217  }
1218 
1219  bool start()
1220  {
1221  bool result = true;
1222  result = writer.start() && result;
1223  result = reader.start() && result;
1224  result = connector.start() && result;
1225  result = another_output_connector.start() && result;
1226  result = another_input_connector.start() && result;
1227  return true;
1228  }
1229 
1230  bool stop()
1231  {
1232  bool result = true;
1233  result = writer.stop() && result;
1234  result = reader.stop() && result;
1235  result = connector.stop() && result;
1236  result = another_output_connector.stop() && result;
1237  result = another_input_connector.stop() && result;
1238  return true;
1239  }
1240 };
1241 
1242 // Registers the fixture into the 'registry'
1243 BOOST_FIXTURE_TEST_SUITE( ConcurrencyPortsTestSuite, ConcurrencyPortsTestFixture )
1244 
1245 BOOST_AUTO_TEST_CASE( testConcurrencyPerConnection )
1246 {
1247  ConnPolicy policy = ConnPolicy::data(ConnPolicy::LOCK_FREE);
1248  policy.buffer_policy = PerConnection;
1249 
1250  BOOST_REQUIRE( connector.connected = connect(policy) );
1251  BOOST_REQUIRE( start() );
1252  sleep(1);
1253  BOOST_REQUIRE( stop() );
1254  disconnect();
1255 
1256  BOOST_TEST_MESSAGE("Number of successful writes: " << writer.write_counter);
1257  BOOST_CHECK_GE( writer.write_counter, 100 );
1258  BOOST_TEST_MESSAGE("Number of successful reads (NewData): " << reader.read_counter);
1259  BOOST_CHECK_GE( reader.read_counter, 100 );
1260  BOOST_TEST_MESSAGE("Number of connects/disconnects: "
1261  << connector.connect_counter << "/" << connector.disconnect_counter
1262  << " (" << connector.connect_failure_counter << "/" << connector.disconnect_failure_counter << " failures)");
1263  BOOST_CHECK_GE( connector.connect_counter, 100 );
1264  BOOST_CHECK_EQUAL( connector.connect_failure_counter, 0 );
1265  BOOST_CHECK_EQUAL( connector.disconnect_failure_counter, 0 );
1266  BOOST_TEST_MESSAGE("Number of connects/disconnects from another output port: "
1267  << another_output_connector.connect_counter << "/" << another_output_connector.disconnect_counter
1268  << " (" << another_output_connector.connect_failure_counter << "/" << another_output_connector.disconnect_failure_counter << " failures)");
1269  BOOST_CHECK_GE( another_output_connector.connect_counter, 100 );
1270  BOOST_CHECK_EQUAL( another_output_connector.connect_failure_counter, 0 );
1271  BOOST_CHECK_EQUAL( another_output_connector.disconnect_failure_counter, 0 );
1272  BOOST_TEST_MESSAGE("Number of connects/disconnects to another input port: "
1273  << another_input_connector.connect_counter << "/" << another_input_connector.disconnect_counter
1274  << " (" << another_input_connector.connect_failure_counter << "/" << another_input_connector.disconnect_failure_counter << " failures)");
1275  BOOST_CHECK_GE( another_input_connector.connect_counter, 100 );
1276  BOOST_CHECK_EQUAL( another_input_connector.connect_failure_counter, 0 );
1277  BOOST_CHECK_EQUAL( another_input_connector.disconnect_failure_counter, 0 );
1278 }
1279 
1280 BOOST_AUTO_TEST_CASE( testConcurrencyPerInputPort )
1281 {
1282  ConnPolicy policy = ConnPolicy::data(ConnPolicy::LOCKED);
1283  policy.buffer_policy = PerInputPort;
1284 
1285  BOOST_REQUIRE( connector.connected = connect(policy) );
1286  BOOST_REQUIRE( start() );
1287  sleep(1);
1288  BOOST_REQUIRE( stop() );
1289  disconnect();
1290 
1291  BOOST_TEST_MESSAGE("Number of successful writes: " << writer.write_counter);
1292  BOOST_CHECK_GE( writer.write_counter, 100 );
1293  BOOST_TEST_MESSAGE("Number of successful reads (NewData): " << reader.read_counter);
1294  BOOST_CHECK_GE( reader.read_counter, 100 );
1295  BOOST_TEST_MESSAGE("Number of connects/disconnects: "
1296  << connector.connect_counter << "/" << connector.disconnect_counter
1297  << " (" << connector.connect_failure_counter << "/" << connector.disconnect_failure_counter << " failures)");
1298  BOOST_CHECK_GE( connector.connect_counter, 100 );
1299  BOOST_CHECK_EQUAL( connector.connect_failure_counter, 0 );
1300  BOOST_CHECK_EQUAL( connector.disconnect_failure_counter, 0 );
1301  BOOST_TEST_MESSAGE("Number of connects/disconnects from another output port: "
1302  << another_output_connector.connect_counter << "/" << another_output_connector.disconnect_counter
1303  << " (" << another_output_connector.connect_failure_counter << "/" << another_output_connector.disconnect_failure_counter << " failures)");
1304  BOOST_CHECK_GE( another_output_connector.connect_counter, 100 );
1305  BOOST_CHECK_EQUAL( another_output_connector.connect_failure_counter, 0 );
1306  BOOST_CHECK_EQUAL( another_output_connector.disconnect_failure_counter, 0 );
1307  BOOST_TEST_MESSAGE("Number of connects/disconnects to another input port: "
1308  << another_input_connector.connect_counter << "/" << another_input_connector.disconnect_counter
1309  << " (" << another_input_connector.connect_failure_counter << "/" << another_input_connector.disconnect_failure_counter << " failures)");
1310  BOOST_CHECK_GE( another_input_connector.connect_counter, 100 );
1311  BOOST_CHECK_EQUAL( another_input_connector.connect_failure_counter, 0 );
1312  BOOST_CHECK_EQUAL( another_input_connector.disconnect_failure_counter, 0 );
1313 }
1314 
1315 BOOST_AUTO_TEST_CASE( testConcurrencyPerOutputPort )
1316 {
1317  ConnPolicy policy = ConnPolicy::data(ConnPolicy::LOCK_FREE);
1318  policy.buffer_policy = PerOutputPort;
1319 
1320  BOOST_REQUIRE( connector.connected = connect(policy) );
1321  BOOST_REQUIRE( start() );
1322  sleep(1);
1323  BOOST_REQUIRE( stop() );
1324  disconnect();
1325 
1326  BOOST_TEST_MESSAGE("Number of successful writes: " << writer.write_counter);
1327  BOOST_CHECK_GE( writer.write_counter, 100 );
1328  BOOST_TEST_MESSAGE("Number of successful reads (NewData): " << reader.read_counter);
1329  BOOST_CHECK_GE( reader.read_counter, 100 );
1330  BOOST_TEST_MESSAGE("Number of connects/disconnects: "
1331  << connector.connect_counter << "/" << connector.disconnect_counter
1332  << " (" << connector.connect_failure_counter << "/" << connector.disconnect_failure_counter << " failures)");
1333  BOOST_CHECK_GE( connector.connect_counter, 100 );
1334  BOOST_CHECK_EQUAL( connector.connect_failure_counter, 0 );
1335  BOOST_CHECK_EQUAL( connector.disconnect_failure_counter, 0 );
1336  BOOST_TEST_MESSAGE("Number of connects/disconnects from another output port: "
1337  << another_output_connector.connect_counter << "/" << another_output_connector.disconnect_counter
1338  << " (" << another_output_connector.connect_failure_counter << "/" << another_output_connector.disconnect_failure_counter << " failures)");
1339  BOOST_CHECK_GE( another_output_connector.connect_counter, 100 );
1340  BOOST_CHECK_EQUAL( another_output_connector.connect_failure_counter, 0 );
1341  BOOST_CHECK_EQUAL( another_output_connector.disconnect_failure_counter, 0 );
1342  BOOST_TEST_MESSAGE("Number of connects/disconnects to another input port: "
1343  << another_input_connector.connect_counter << "/" << another_input_connector.disconnect_counter
1344  << " (" << another_input_connector.connect_failure_counter << "/" << another_input_connector.disconnect_failure_counter << " failures)");
1345  BOOST_CHECK_GE( another_input_connector.connect_counter, 100 );
1346  BOOST_CHECK_EQUAL( another_input_connector.connect_failure_counter, 0 );
1347  BOOST_CHECK_EQUAL( another_input_connector.disconnect_failure_counter, 0 );
1348 }
1349 
1350 BOOST_AUTO_TEST_CASE( testConcurrencySharedConnection )
1351 {
1352  ConnPolicy policy = ConnPolicy::data(ConnPolicy::LOCKED);
1353  policy.buffer_policy = Shared;
1354 
1355  BOOST_REQUIRE( connector.connected = connect(policy) );
1356  connector.shared_connection = writer.port.getSharedConnection();
1357  another_output_connector.shared_connection = writer.port.getSharedConnection();
1358  another_input_connector.shared_connection = reader.port.getSharedConnection();
1359 
1360  // Add another fake writer connection so that the shared connection is not cleaned up if writer.port and another_output_port disconnect.
1361  RTT::OutputPort<T> yet_another_output_port("yet_another_output_port");
1362  yet_another_output_port.createConnection(writer.port.getSharedConnection(), policy);
1363 
1364  BOOST_REQUIRE( start() );
1365  sleep(1);
1366  BOOST_REQUIRE( stop() );
1367  disconnect();
1368 
1369  BOOST_TEST_MESSAGE("Number of successful writes: " << writer.write_counter);
1370  BOOST_CHECK_GE( writer.write_counter, 100 );
1371  BOOST_TEST_MESSAGE("Number of successful reads (NewData): " << reader.read_counter);
1372  BOOST_CHECK_GE( reader.read_counter, 100 );
1373  BOOST_TEST_MESSAGE("Number of connects/disconnects: "
1374  << connector.connect_counter << "/" << connector.disconnect_counter
1375  << " (" << connector.connect_failure_counter << "/" << connector.disconnect_failure_counter << " failures)");
1376  BOOST_CHECK_GE( connector.connect_counter, 100 );
1377  BOOST_CHECK_EQUAL( connector.connect_failure_counter, 0 );
1378  BOOST_CHECK_EQUAL( connector.disconnect_failure_counter, 0 );
1379  BOOST_TEST_MESSAGE("Number of connects/disconnects from another output port: "
1380  << another_output_connector.connect_counter << "/" << another_output_connector.disconnect_counter
1381  << " (" << another_output_connector.connect_failure_counter << "/" << another_output_connector.disconnect_failure_counter << " failures)");
1382  BOOST_CHECK_GE( another_output_connector.connect_counter, 100 );
1383  BOOST_CHECK_EQUAL( another_output_connector.connect_failure_counter, 0 );
1384  BOOST_CHECK_EQUAL( another_output_connector.disconnect_failure_counter, 0 );
1385  BOOST_TEST_MESSAGE("Number of connects/disconnects to another input port: "
1386  << another_input_connector.connect_counter << "/" << another_input_connector.disconnect_counter
1387  << " (" << another_input_connector.connect_failure_counter << "/" << another_input_connector.disconnect_failure_counter << " failures)");
1388  BOOST_CHECK_GE( another_input_connector.connect_counter, 100 );
1389  BOOST_CHECK_EQUAL( another_input_connector.connect_failure_counter, 0 );
1390  BOOST_CHECK_EQUAL( another_input_connector.disconnect_failure_counter, 0 );
1391 }
1392 
1393 BOOST_AUTO_TEST_SUITE_END()
RTT::OutputPort::setDataSample
void setDataSample(const T &sample)
Definition: OutputPort.hpp:209
RTT::ConnPolicy::data
static ConnPolicy data(int lock_policy=LOCK_FREE, bool init_connection=true, bool pull=false)
Definition: ConnPolicy.cpp:99
PortsTestFixture::tc
TaskContext * tc
Definition: ports_test.cpp:64
BOOST_FIXTURE_TEST_SUITE
#define BOOST_FIXTURE_TEST_SUITE(suite_name, F)
Definition: unit_test_suite.hpp:53
RTT::internal::SharedConnectionBase::shared_ptr
boost::intrusive_ptr< SharedConnectionBase > shared_ptr
Definition: SharedConnection.hpp:75
RTT::Activity::stop
virtual bool stop()
Definition: Activity.cpp:280
PortsTestFixture::stsim
ActivityInterface * stsim
Definition: ports_test.cpp:69
ConcurrencyPortsTestFixture::connect
bool connect(const ConnPolicy &policy)
Definition: ports_test.cpp:1205
PortsTestFixture::tc3
EventPortsTC * tc3
Definition: ports_test.cpp:67
RTT::extras::SequentialActivity
The default, thread-less activity for any newly created TaskContext.
Definition: SequentialActivity.hpp:71
PerConnection
Definition: dataflow_performance_test.cpp:297
Shared
Definition: dataflow_performance_test.cpp:297
PerOutputPort
Definition: dataflow_performance_test.cpp:297
WriteSuccess
Definition: dataflow_performance_test.cpp:296
BOOST_AUTO_TEST_SUITE_END
#define BOOST_AUTO_TEST_SUITE_END()
Definition: unit_test_suite.hpp:62
EventPortsTC::resetStats
void resetStats()
Definition: ports_test.cpp:52
RTT::InputPort::getSharedBuffer
virtual base::ChannelElement< T >::shared_ptr getSharedBuffer() const
Definition: InputPort.hpp:235
ConcurrencyPortsTestFixture::writer
PortWriterThread< T > writer
Definition: ports_test.cpp:1181
std
Definition: mystd.hpp:163
RTT::base::PortInterface::getName
const std::string & getName() const
Definition: PortInterface.hpp:86
PortsTestFixture::new_data_listener
void new_data_listener(PortInterface *port)
Definition: ports_test.cpp:73
PortsTestFixture::tce
EventPortsTC * tce
Definition: ports_test.cpp:65
ConcurrencyPortsTestFixture::PortConnectorThread::shared_connection
RTT::internal::SharedConnectionBase::shared_ptr shared_connection
Definition: ports_test.cpp:1172
PortsTestFixture::tsim
ActivityInterface * tsim
Definition: ports_test.cpp:68
RTT::ConnPolicy::LOCKED
static const int LOCKED
Definition: ConnPolicy.hpp:116
RTT::InputPort::read
FlowStatus read(base::DataSourceBase::shared_ptr source)
Definition: InputPort.hpp:97
PerInputPort
Definition: dataflow_performance_test.cpp:297
RTT::TaskContext::setActivity
bool setActivity(base::ActivityInterface *new_act)
Definition: TaskContext.cpp:340
PortsTestFixture::slsim
ActivityInterface * slsim
Definition: ports_test.cpp:70
RTT::base::PortInterface::getInterface
DataFlowInterface * getInterface() const
Definition: PortInterface.cpp:117
NotConnected
Definition: dataflow_performance_test.cpp:296
WriteFailure
Definition: dataflow_performance_test.cpp:296
RTT::Activity::start
virtual bool start()
Definition: Activity.cpp:275
RTT::base::OutputPortInterface::connectTo
virtual bool connectTo(PortInterface *other, ConnPolicy const &policy)
Definition: OutputPortInterface.cpp:101
RTT::Handle::connect
bool connect()
Definition: Handle.cpp:65
RTT::ConnPolicy::buffer
static ConnPolicy buffer(int size, int lock_policy=LOCK_FREE, bool init_connection=false, bool pull=false)
Definition: ConnPolicy.cpp:77
ConcurrencyPortsTestFixture::reader
PortReaderThread< T > reader
Definition: ports_test.cpp:1182
RTT::OutputPort::write
WriteStatus write(const T &sample)
Definition: OutputPort.hpp:243
RTT::ConnPolicy::LOCK_FREE
static const int LOCK_FREE
Definition: ConnPolicy.hpp:117
ConcurrencyPortsTestFixture::connector
PortConnectorThread connector
Definition: ports_test.cpp:1183
ConcurrencyPortsTestFixture::another_output_port
OutputPort< T > another_output_port
Definition: ports_test.cpp:1184
RTT::Activity
An Activity executes a RunnableInterface object in a (periodic) thread.
Definition: Activity.hpp:70
sleep
unsigned int sleep(unsigned int s)
Definition: fosi.cpp:51
adaptor::yield
static void yield(TaskContext *tc)
Definition: dataflow_performance_test.cpp:338
PortsTestFixture::signalled_port
PortInterface * signalled_port
Definition: ports_test.cpp:72
PortsTestFixture::tc2
EventPortsTC * tc2
Definition: ports_test.cpp:66
ConcurrencyPortsTestFixture::another_output_connector
PortConnectorThread another_output_connector
Definition: ports_test.cpp:1186
RTT
Contains TaskContext, Activity, OperationCaller, Operation, Property, InputPort, OutputPort, Attribute.
Definition: Activity.cpp:53
ConcurrencyPortsTestFixture::PortConnectorThread::PortConnectorThread
PortConnectorThread(PortInterface &connect_port, PortInterface &other_port, const ConnPolicy &policy=ConnPolicy())
Definition: ports_test.cpp:1141
RTT::extras::SimulationActivity
A SimulationActivity is a PeriodicActivity which is used for simulation.
Definition: SimulationActivity.hpp:59
RTT::extras::SlaveActivity
An base::ActivityInterface implementation which executes &#39;step&#39; upon the invocation of &#39;execute()&#39;...
Definition: SlaveActivity.hpp:90
RTT::OutputPort::getLastWrittenValue
T getLastWrittenValue() const
Definition: OutputPort.hpp:173
RTT::base::OutputPortInterface::createBufferConnection
bool createBufferConnection(InputPortInterface &sink, int size, int lock_policy=ConnPolicy::LOCK_FREE)
Definition: OutputPortInterface.cpp:90
EventPortsTC::updateHook
void updateHook()
Definition: ports_test.cpp:47
RTT::Handle
The Handle holds the information, and allows manipulation, of a connection between a internal::Signal...
Definition: Handle.hpp:66
BOOST_AUTO_TEST_CASE
BOOST_AUTO_TEST_CASE(testPortTaskInterface)
Definition: ports_test.cpp:111
RTT::OutputPort::antiClone
virtual base::PortInterface * antiClone() const
Definition: OutputPort.hpp:302
RTT::TaskContext::getActivity
base::ActivityInterface * getActivity()
Definition: TaskContext.cpp:390
RTT::InputPort::getDataSample
void getDataSample(T &sample)
Definition: InputPort.hpp:176
RTT::OutputPort::createConnection
virtual bool createConnection(base::InputPortInterface &input_port, ConnPolicy const &policy)
Definition: OutputPort.hpp:309
ConcurrencyPortsTestFixture::another_input_port
InputPort< T > another_input_port
Definition: ports_test.cpp:1185
adaptor::trigger
static bool trigger(TaskContext *tc)
Definition: dataflow_performance_test.cpp:334

rtt
Author(s): RTT Developers
autogenerated on Tue Jun 25 2019 19:33:26