shadow_mcevent.c

 
#include <freerdp/config.h>
 
#include <winpr/assert.h>
#include <freerdp/log.h>
#include "shadow.h"
 
#define TAG SERVER_TAG("shadow.mcevent")
 
/* Maximum time the publisher waits for all subscribers to consume a frame.
 * Without this bound a single subscriber that stops consuming (e.g. a client
 * stuck before its TLS handshake, or one stalled mid-send) blocks the
 * subsystem/capture thread forever and, on the Windows subsystem, the
 * listener thread with it (the clients list lock is held across the wait). */
#define SHADOW_SUBSCRIBER_WAIT_TIMEOUT 1000ul
 
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 = calloc(1, sizeof(rdpShadowMultiClientEvent));
  if (!event)
    goto fail;
 
  if (!InitializeCriticalSectionAndSpinCount(&(event->lock), 4000))
    goto fail;
 
  event->event = CreateEvent(nullptr, TRUE, FALSE, nullptr);
  if (!event->event)
    goto fail;
 
  event->barrierEvent = CreateEvent(nullptr, TRUE, FALSE, nullptr);
  if (!event->barrierEvent)
    goto fail;
 
  event->doneEvent = CreateEvent(nullptr, TRUE, FALSE, nullptr);
  if (!event->doneEvent)
    goto fail;
 
  event->subscribers = ArrayList_New(TRUE);
  if (!event->subscribers)
    goto fail;
 
  event->consuming = 0;
  event->waiting = 0;
  event->eventid = 0;
  if (!SetEvent(event->doneEvent))
    goto fail;
  return event;
 
fail:
  shadow_multiclient_free(event);
  return nullptr;
}
 
void shadow_multiclient_free(rdpShadowMultiClientEvent* event)
{
  if (!event)
    return;
 
  DeleteCriticalSection(&(event->lock));
 
  ArrayList_Free(event->subscribers);
  if (event->doneEvent)
    (void)CloseHandle(event->doneEvent);
  if (event->barrierEvent)
    (void)CloseHandle(event->barrierEvent);
  if (event->event)
    (void)CloseHandle(event->event);
  free(event);
}
 
static void Publish(rdpShadowMultiClientEvent* event)
{
  WINPR_ASSERT(event);
  wArrayList* subscribers = event->subscribers;
 
  WINPR_ASSERT(event->consuming == 0);
 
  /* Count subscribing clients */
  ArrayList_Lock(subscribers);
  for (size_t i = 0; i < ArrayList_Count(subscribers); i++)
  {
    struct rdp_shadow_multiclient_subscriber* 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 DiscardRound(rdpShadowMultiClientEvent* event)
{
  wArrayList* subscribers = nullptr;
 
  WINPR_ASSERT(event);
 
  /* Called with event->lock held after the subscriber wait timed out.
   * Reset the round state so that the publisher can continue: subscribers
   * waking up late see pleaseHandle == FALSE and skip their consume, and
   * clients blocked on the barrier event are released. */
  subscribers = event->subscribers;
  ArrayList_Lock(subscribers);
  for (size_t i = 0; i < ArrayList_Count(subscribers); i++)
  {
    struct rdp_shadow_multiclient_subscriber* subscriber =
        (struct rdp_shadow_multiclient_subscriber*)ArrayList_GetItem(subscribers, i);
    subscriber->pleaseHandle = FALSE;
  }
  ArrayList_Unlock(subscribers);
 
  if (event->waiting > 0)
  {
    (void)SetEvent(event->barrierEvent);
  }
  else
  {
    (void)SetEvent(event->doneEvent);
  }
 
  WLog_WARN(TAG, "Event %d dropped: %d subscriber(s) unresponsive for %lu ms", event->eventid,
            event->consuming, SHADOW_SUBSCRIBER_WAIT_TIMEOUT);
  event->consuming = 0;
  (void)ResetEvent(event->event);
}
 
static void WaitForSubscribers(rdpShadowMultiClientEvent* event)
{
  DWORD status = WAIT_FAILED;
 
  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));
    status = WaitForSingleObject(event->doneEvent, SHADOW_SUBSCRIBER_WAIT_TIMEOUT);
    EnterCriticalSection(&(event->lock));
 
    if (status != WAIT_OBJECT_0)
      DiscardRound(event);
 
    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);
 
        /* a round dropped while we waited may be followed by a new one already:
         * its own last client tells the server */
        if (event->consuming == 0)
          (void)SetEvent(event->doneEvent);
      }
    }
  }
 
  return ret;
}
 
void* shadow_multiclient_get_subscriber(rdpShadowMultiClientEvent* event)
{
  struct rdp_shadow_multiclient_subscriber* subscriber = nullptr;
 
  if (!event)
    return nullptr;
 
  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 nullptr;
}
 
/*
 * 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 = nullptr;
  rdpShadowMultiClientEvent* event = nullptr;
 
  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 = nullptr;
  rdpShadowMultiClientEvent* event = nullptr;
  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) nullptr;
 
  return ((struct rdp_shadow_multiclient_subscriber*)subscriber)->ref->event;
}