shadow_mcevent.c

 
#include <freerdp/config.h>
 
#include <winpr/assert.h>
#include <freerdp/log.h>
#include "shadow.h"
 
#define TAG SERVER_TAG("shadow.mcevent")
 
struct rdp_shadow_multiclient_event
{
  HANDLE event;        /* Kickoff event */
  HANDLE barrierEvent; /* Represents that all clients have consumed event */
  HANDLE doneEvent;    /* Event handling finished. Server could continue */
  wArrayList* subscribers;
  CRITICAL_SECTION lock;
  int consuming;
  int waiting;
 
  /* For debug */
  int eventid;
};
 
struct rdp_shadow_multiclient_subscriber
{
  rdpShadowMultiClientEvent* ref;
  BOOL pleaseHandle; /* Indicate if server expects my handling in this turn */
};
 
rdpShadowMultiClientEvent* shadow_multiclient_new(void)
{
  rdpShadowMultiClientEvent* event =
      (rdpShadowMultiClientEvent*)calloc(1, sizeof(rdpShadowMultiClientEvent));
  if (!event)
    goto out_error;
 
  event->event = CreateEvent(NULL, TRUE, FALSE, NULL);
  if (!event->event)
    goto out_free;
 
  event->barrierEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
  if (!event->barrierEvent)
    goto out_free_event;
 
  event->doneEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
  if (!event->doneEvent)
    goto out_free_barrierEvent;
 
  event->subscribers = ArrayList_New(TRUE);
  if (!event->subscribers)
    goto out_free_doneEvent;
 
  if (!InitializeCriticalSectionAndSpinCount(&(event->lock), 4000))
    goto out_free_subscribers;
 
  event->consuming = 0;
  event->waiting = 0;
  event->eventid = 0;
  (void)SetEvent(event->doneEvent);
  return event;
 
out_free_subscribers:
  ArrayList_Free(event->subscribers);
out_free_doneEvent:
  (void)CloseHandle(event->doneEvent);
out_free_barrierEvent:
  (void)CloseHandle(event->barrierEvent);
out_free_event:
  (void)CloseHandle(event->event);
out_free:
  free(event);
out_error:
  return (rdpShadowMultiClientEvent*)NULL;
}
 
void shadow_multiclient_free(rdpShadowMultiClientEvent* event)
{
  if (!event)
    return;
 
  DeleteCriticalSection(&(event->lock));
 
  ArrayList_Free(event->subscribers);
  (void)CloseHandle(event->doneEvent);
  (void)CloseHandle(event->barrierEvent);
  (void)CloseHandle(event->event);
  free(event);
}
 
static void Publish(rdpShadowMultiClientEvent* event)
{
  wArrayList* subscribers = NULL;
  struct rdp_shadow_multiclient_subscriber* subscriber = NULL;
 
  subscribers = event->subscribers;
 
  WINPR_ASSERT(event->consuming == 0);
 
  /* Count subscribing clients */
  ArrayList_Lock(subscribers);
  for (size_t i = 0; i < ArrayList_Count(subscribers); i++)
  {
    subscriber = (struct rdp_shadow_multiclient_subscriber*)ArrayList_GetItem(subscribers, i);
    /* Set flag to subscriber: I acknowledge and please handle */
    subscriber->pleaseHandle = TRUE;
    event->consuming++;
  }
  ArrayList_Unlock(subscribers);
 
  if (event->consuming > 0)
  {
    event->eventid = (event->eventid & 0xff) + 1;
    WLog_VRB(TAG, "Server published event %d. %d clients.\n", event->eventid, event->consuming);
    (void)ResetEvent(event->doneEvent);
    (void)SetEvent(event->event);
  }
}
 
static void WaitForSubscribers(rdpShadowMultiClientEvent* event)
{
  if (event->consuming > 0)
  {
    /* Wait for clients done */
    WLog_VRB(TAG, "Server wait event %d. %d clients.\n", event->eventid, event->consuming);
    LeaveCriticalSection(&(event->lock));
    (void)WaitForSingleObject(event->doneEvent, INFINITE);
    EnterCriticalSection(&(event->lock));
    WLog_VRB(TAG, "Server quit event %d. %d clients.\n", event->eventid, event->consuming);
  }
 
  /* Last subscriber should have already reset the event */
  WINPR_ASSERT(WaitForSingleObject(event->event, 0) != WAIT_OBJECT_0);
}
 
void shadow_multiclient_publish(rdpShadowMultiClientEvent* event)
{
  if (!event)
    return;
 
  EnterCriticalSection(&(event->lock));
  Publish(event);
  LeaveCriticalSection(&(event->lock));
}
void shadow_multiclient_wait(rdpShadowMultiClientEvent* event)
{
  if (!event)
    return;
 
  EnterCriticalSection(&(event->lock));
  WaitForSubscribers(event);
  LeaveCriticalSection(&(event->lock));
}
void shadow_multiclient_publish_and_wait(rdpShadowMultiClientEvent* event)
{
  if (!event)
    return;
 
  EnterCriticalSection(&(event->lock));
  Publish(event);
  WaitForSubscribers(event);
  LeaveCriticalSection(&(event->lock));
}
 
static BOOL Consume(struct rdp_shadow_multiclient_subscriber* subscriber, BOOL wait)
{
  rdpShadowMultiClientEvent* event = subscriber->ref;
  BOOL ret = FALSE;
 
  if (WaitForSingleObject(event->event, 0) == WAIT_OBJECT_0 && subscriber->pleaseHandle)
  {
    /* Consume my share. Server is waiting for us */
    event->consuming--;
    ret = TRUE;
  }
 
  WINPR_ASSERT(event->consuming >= 0);
 
  if (event->consuming == 0)
  {
    /* Last client reset event before notify clients to continue */
    (void)ResetEvent(event->event);
 
    if (event->waiting > 0)
    {
      /* Notify other clients to continue */
      (void)SetEvent(event->barrierEvent);
    }
    else
    {
      /* Only one client. Notify server directly */
      (void)SetEvent(event->doneEvent);
    }
  }
  else /* (event->consuming > 0) */
  {
    if (wait)
    {
      /*
       * This client need to wait. That means the client will
       * continue waiting for other clients to finish.
       * The last client should reset barrierEvent.
       */
      event->waiting++;
      LeaveCriticalSection(&(event->lock));
      (void)WaitForSingleObject(event->barrierEvent, INFINITE);
      EnterCriticalSection(&(event->lock));
      event->waiting--;
      if (event->waiting == 0)
      {
        /*
         * This is last client waiting for barrierEvent.
         * We can now discard barrierEvent and notify
         * server to continue.
         */
        (void)ResetEvent(event->barrierEvent);
        (void)SetEvent(event->doneEvent);
      }
    }
  }
 
  return ret;
}
 
void* shadow_multiclient_get_subscriber(rdpShadowMultiClientEvent* event)
{
  struct rdp_shadow_multiclient_subscriber* subscriber = NULL;
 
  if (!event)
    return NULL;
 
  EnterCriticalSection(&(event->lock));
 
  subscriber = (struct rdp_shadow_multiclient_subscriber*)calloc(
      1, sizeof(struct rdp_shadow_multiclient_subscriber));
  if (!subscriber)
    goto out_error;
 
  subscriber->ref = event;
  subscriber->pleaseHandle = FALSE;
 
  if (!ArrayList_Append(event->subscribers, subscriber))
    goto out_free;
 
  WLog_VRB(TAG, "Get subscriber %p. Wait event %d. %d clients.\n", (void*)subscriber,
           event->eventid, event->consuming);
  (void)Consume(subscriber, TRUE);
  WLog_VRB(TAG, "Get subscriber %p. Quit event %d. %d clients.\n", (void*)subscriber,
           event->eventid, event->consuming);
 
  LeaveCriticalSection(&(event->lock));
 
  return subscriber;
 
out_free:
  free(subscriber);
out_error:
  LeaveCriticalSection(&(event->lock));
  return NULL;
}
 
/*
 * Consume my share and release my register
 * If we have update event and pleaseHandle flag
 * We need to consume. Anyway we need to clear
 * pleaseHandle flag
 */
void shadow_multiclient_release_subscriber(void* subscriber)
{
  struct rdp_shadow_multiclient_subscriber* s = NULL;
  rdpShadowMultiClientEvent* event = NULL;
 
  if (!subscriber)
    return;
 
  s = (struct rdp_shadow_multiclient_subscriber*)subscriber;
  event = s->ref;
 
  EnterCriticalSection(&(event->lock));
 
  WLog_VRB(TAG, "Release Subscriber %p. Drop event %d. %d clients.\n", subscriber, event->eventid,
           event->consuming);
  (void)Consume(s, FALSE);
  WLog_VRB(TAG, "Release Subscriber %p. Quit event %d. %d clients.\n", subscriber, event->eventid,
           event->consuming);
 
  ArrayList_Remove(event->subscribers, subscriber);
 
  LeaveCriticalSection(&(event->lock));
 
  free(subscriber);
}
 
BOOL shadow_multiclient_consume(void* subscriber)
{
  struct rdp_shadow_multiclient_subscriber* s = NULL;
  rdpShadowMultiClientEvent* event = NULL;
  BOOL ret = FALSE;
 
  if (!subscriber)
    return ret;
 
  s = (struct rdp_shadow_multiclient_subscriber*)subscriber;
  event = s->ref;
 
  EnterCriticalSection(&(event->lock));
 
  WLog_VRB(TAG, "Subscriber %p wait event %d. %d clients.\n", subscriber, event->eventid,
           event->consuming);
  ret = Consume(s, TRUE);
  WLog_VRB(TAG, "Subscriber %p quit event %d. %d clients.\n", subscriber, event->eventid,
           event->consuming);
 
  LeaveCriticalSection(&(event->lock));
 
  return ret;
}
 
HANDLE shadow_multiclient_getevent(void* subscriber)
{
  if (!subscriber)
    return (HANDLE)NULL;
 
  return ((struct rdp_shadow_multiclient_subscriber*)subscriber)->ref->event;
}