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

rtt
Author(s): RTT Developers
autogenerated on Fri Oct 25 2019 03:59:34