Ref: synchronizing init task with an arbitrary number of worker tasks

[Andrei Dimitrief-Jianu]

>On Wed, Sep 26, 2012 at 9:05 AM, Joel Sherrill <joel.sherrill at oarcorp.com> wrote:
>> On 09/25/2012 09:22 PM, Andrei Dimitrief-Jianu wrote:
>>
>> Yes, I could use a barrier w/ RTEMS_BARRIER_AUTOMATIC_RELEASE.
>> However, I would prevent from finishing the tasks that I want to
>> synchronize (probably this is the reason why barriers are recommended
>> for the initialization phase). I need something that would let the
>> workers to finish and keep a count of how many of them have finished,
>> and when that count is reached it would release the init task as well.
>>
>> What I was looking for was a synchronization event that could be
>> described by an API similar to:
>>
>> rtems_status_code rtems_synch_event_initialize( rtems_synch_event
>> *synch_event, int wait_count );
>>
>> rtems_status_code rtems_synch_event_wait_one( rtems_synch_event
>> *synch_event );
>>
>> rtems_status_code rtems_synch_event_wait_all( rtems_synch_event
>> *synch_event );
>>
>> rtems_status_code rtems_synch_event_set( rtems_synch_event *synch_event );
>>
>> rtems_status_code rtems_synch_event_reset( rtems_synch_event *synch_event
>> );
>
> I think you are wanting an event set that is independent of a particular
> task. It is an independent object. I have implemented this type of object
> for some non-RTEMS customer specific middleware. It is quite complicated
> to get the semantics correct.
>
> + If you allow posting an event to wake up multiple tasks, then it is O(n)
>     when you post an event.
> + There is the issue of whether you allow starvation or scan the entire
>     blocking set of tasks to see who might be satisfied.
> + When do events get cleared? When the first blocked task is satisfied
>     or when all tasks who might be interested in this task are unblocked?
>
> I would be happy to see something like this implemented in RTEMS and
> have wanted an event manager that is independent of tasks. But getting
> the definition and behaviour correct is not simple.
>
> Another design alternative is to use the classic API event manager
> for up to 32 tasks. Assign an event flag (e.g. bit) to each task. The init
> task will wait for all tasks to post their event (w/ a timeout I hope). The
> application tasks should post their assigned bit when they completed
> initialization.  That lets the initialization task know which tasks have
> completed initialization within the allotted time.
>
> To hold the application tasks off until all are finished, they should post
> their event flag to the Init task. Then block on a manual release barrier.
> The application tasks should need to be non-preemptible during
> initialization
> to ensure they post their flag and block atomically.
>
>    /* created as non-preemptible, initialization priority */
>    rtems_event_send( Init_task_id, my_event_flag );
>    rtems_barrier_wait( Init_barrier, forever );
>        /* entire application is up, not set our mode/priority to "normal"
> values */
>    /* optional */  rtems_task_set_priority( RTEMS_SELF, ... desired run-time
> priority )
>    rtems_task_mode( RTEMS_SELF, args for desired preemption mode )
>
> To be completely paranoid, the initialization task could check that the
> count
> at the barrier matches the expectations before releasing them. This gives
> you the detailed information that the global variables solution does and a
> timeout
> capability if all tasks do not complete initialization. If you fail to get
> through
> all the application tasks' initialization, then it should be easy to report
> who
> failed.
>
> --joel
>


There are definitely many ways of getting it done. I have used a
semaphore to synchronize tasks access to the event count variable
(hence the need for a delete method in order to release the resources
used by the semaphore).

I have below an example for the synch_event used with three worker
tasks. It is not yet implemented as a RTEMS manager, but gets the job
done.

Regards,
Andrei.



#include <bsp.h>

#include <stdlib.h>
#include <stdio.h>

/////////////////////////////////////////////////////////////
struct rtems_synch_event
{
  uint16_t event_count;
  rtems_id semaphore_id;
};
rtems_status_code rtems_synch_event_initialize( struct
rtems_synch_event *synch_event, uint16_t event_count );
rtems_status_code rtems_synch_event_delete( struct rtems_synch_event
*synch_event );
rtems_status_code rtems_synch_event_count( struct rtems_synch_event
*synch_event, uint16_t *event_count );
rtems_status_code rtems_synch_event_wait_one( struct rtems_synch_event
*synch_event );
rtems_status_code rtems_synch_event_wait_all( struct rtems_synch_event
*synch_event );
rtems_status_code rtems_synch_event_set( struct rtems_synch_event
*synch_event );
/////////////////////////////////////////////////////////////

#define WORKER_TASKS 3

rtems_task worker_task( rtems_task_argument argument );

struct rtems_synch_event synch_event;

rtems_task
Init( rtems_task_argument ignored )
{
  rtems_id  task_id;
  rtems_status_code status_code;
  rtems_name task_name;

  status_code = rtems_synch_event_initialize( &synch_event, WORKER_TASKS );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_synch_event_initialize failed with status of
%d.\n", status_code );
    exit( 1 );
  }

  /////////////////////////////////////////////////////////////////////////////
  task_name = rtems_build_name( 'T', 'S', 'K', '1' );

  status_code = rtems_task_create(
    task_name, 1, RTEMS_MINIMUM_STACK_SIZE,
    RTEMS_NO_PREEMPT, RTEMS_FLOATING_POINT, &task_id );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_task_create failed with status of %d.\n", status_code );
    exit( 1 );
  }

  status_code = rtems_task_start( task_id, worker_task, 1 );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_task_start failed with status of %d.\n", status_code );
    exit( 1 );
  }
  /////////////////////////////////////////////////////////////////////////////
  task_name = rtems_build_name( 'T', 'S', 'K', '2' );

  status_code = rtems_task_create(
    task_name, 1, RTEMS_MINIMUM_STACK_SIZE,
    RTEMS_NO_PREEMPT, RTEMS_FLOATING_POINT, &task_id );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_task_create failed with status of %d.\n", status_code );
    exit( 1 );
  }

  status_code = rtems_task_start( task_id, worker_task, 2 );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_task_start failed with status of %d.\n", status_code );
    exit( 1 );
  }
  /////////////////////////////////////////////////////////////////////////////
  task_name = rtems_build_name( 'T', 'S', 'K', '3' );

  status_code = rtems_task_create(
    task_name, 1, RTEMS_MINIMUM_STACK_SIZE,
    RTEMS_NO_PREEMPT, RTEMS_FLOATING_POINT, &task_id );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_task_create failed with status of %d.\n", status_code );
    exit( 1 );
  }

  status_code = rtems_task_start( task_id, worker_task, 3 );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_task_start failed with status of %d.\n", status_code );
    exit( 1 );
  }
  /////////////////////////////////////////////////////////////////////////////

  printf( "\nmain task waiting for the workers to finish...\n" );
  rtems_synch_event_wait_all( &synch_event );

  status_code = rtems_synch_event_delete( &synch_event );
  if( RTEMS_SUCCESSFUL != status_code )
  {
    printf( "\nrtems_synch_event_delete failed with status of %d.\n",
status_code );
    exit( 1 );
  }

  printf( "\nmain task is exiting...\n" );

  exit( 0 );
}

rtems_task
worker_task( rtems_task_argument argument )
{
  int index = 0;
  while( index++ < 10 * argument )
  {
    printf( "\ntask [%d] > doing something in the user task [%d].\n",
(uint16_t)argument, index );
    rtems_task_wake_after( RTEMS_MILLISECONDS_TO_TICKS( 1000 ) );
  }

  rtems_synch_event_set( &synch_event );

  printf( "\ntask [%d] is exiting...\n", (uint16_t)argument );
  rtems_task_delete( rtems_task_self() );

}


#define CONFIGURE_APPLICATION_NEEDS_CONSOLE_DRIVER

#define CONFIGURE_APPLICATION_NEEDS_CLOCK_DRIVER
#define CONFIGURE_MICROSECONDS_PER_TICK 1000
#define CONFIGURE_TICKS_PER_TIMESLICE 50

#define CONFIGURE_MAXIMUM_TASKS 4

#define CONFIGURE_RTEMS_INIT_TASKS_TABLE

#define CONFIGURE_INIT

#include <rtems/confdefs.h>


/////////////////////////////////////////////////////////////
rtems_status_code
rtems_synch_event_initialize( struct rtems_synch_event *synch_event,
uint16_t event_count )
{
  rtems_status_code status_code = RTEMS_SUCCESSFUL;

  synch_event->event_count = event_count;
  status_code = rtems_semaphore_create(
    rtems_build_name( 'S', 'Y', 'N' ,'C' ),
    1,
    RTEMS_LOCAL | RTEMS_FIFO | RTEMS_SIMPLE_BINARY_SEMAPHORE,
    RTEMS_LOCAL | RTEMS_NO_INHERIT_PRIORITY,
    &synch_event->semaphore_id
    );

  return status_code;
}

rtems_status_code
rtems_synch_event_delete( struct rtems_synch_event *synch_event )
{
  rtems_status_code status_code = RTEMS_SUCCESSFUL;

  synch_event->event_count = 0;
  status_code = rtems_semaphore_delete( synch_event->semaphore_id );

  return status_code;
}

rtems_status_code
rtems_synch_event_count( struct rtems_synch_event *synch_event,
uint16_t *event_count )
{
  rtems_status_code status_code = RTEMS_SUCCESSFUL;

  status_code = rtems_semaphore_obtain( synch_event->semaphore_id,
RTEMS_WAIT, RTEMS_NO_TIMEOUT );
  if( RTEMS_SUCCESSFUL == status_code )
  {
    *event_count = synch_event->event_count;
    status_code = rtems_semaphore_release( synch_event->semaphore_id );
  }

  return status_code;
}

rtems_status_code
rtems_synch_event_wait_one( struct rtems_synch_event *synch_event )
{
  rtems_status_code status_code = RTEMS_SUCCESSFUL;

  uint16_t initial_event_count;
  status_code = rtems_synch_event_count( synch_event, &initial_event_count );
  if( RTEMS_SUCCESSFUL == status_code )
  {
    uint16_t event_count;
    do
    {
      status_code = rtems_semaphore_obtain( synch_event->semaphore_id,
RTEMS_WAIT, RTEMS_NO_TIMEOUT );
      if( RTEMS_SUCCESSFUL == status_code )
      {
        event_count = synch_event->event_count;
        status_code = rtems_semaphore_release( synch_event->semaphore_id );
      }
      rtems_task_wake_after( CONFIGURE_TICKS_PER_TIMESLICE );
    }
    while( RTEMS_SUCCESSFUL == status_code && initial_event_count - 1
< event_count );
  }

  return status_code;
}

rtems_status_code
rtems_synch_event_wait_all( struct rtems_synch_event *synch_event )
{
  rtems_status_code status_code = RTEMS_SUCCESSFUL;

  uint16_t event_count;
  do
  {
    status_code = rtems_semaphore_obtain( synch_event->semaphore_id,
RTEMS_WAIT, RTEMS_NO_TIMEOUT );
    if( RTEMS_SUCCESSFUL == status_code )
    {
      event_count = synch_event->event_count;
      status_code = rtems_semaphore_release( synch_event->semaphore_id );
    }
    rtems_task_wake_after( CONFIGURE_TICKS_PER_TIMESLICE );
  }
  while( RTEMS_SUCCESSFUL == status_code && 0 < event_count );

  return status_code;
}

rtems_status_code
rtems_synch_event_set( struct rtems_synch_event *synch_event )
{
  rtems_status_code status_code = RTEMS_SUCCESSFUL;

  status_code = rtems_semaphore_obtain( synch_event->semaphore_id,
RTEMS_WAIT, RTEMS_NO_TIMEOUT );
  if( RTEMS_SUCCESSFUL == status_code)
  {
    synch_event->event_count--;
    status_code = rtems_semaphore_release( synch_event->semaphore_id );
  }

  return status_code;
}
/////////////////////////////////////////////////////////////