Merge pull request #2557 from realjiangms/shadow_fix_hang2395

shadow fix multi-client hang

include/freerdp/server/shadow.h

@@ -46,6 +46,7 @@ typedef struct rdp_shadow_surface rdpShadowSurface;
 typedef struct rdp_shadow_encoder rdpShadowEncoder;
 typedef struct rdp_shadow_capture rdpShadowCapture;
 typedef struct rdp_shadow_subsystem rdpShadowSubsystem;
+typedef struct rdp_shadow_multiclient_event rdpShadowMultiClientEvent;
 
 typedef struct _RDP_SHADOW_ENTRY_POINTS RDP_SHADOW_ENTRY_POINTS;
 typedef int (*pfnShadowSubsystemEntry)(RDP_SHADOW_ENTRY_POINTS* pEntryPoints);
@@ -143,11 +144,10 @@ struct _RDP_SHADOW_ENTRY_POINTS
 	int selectedMonitor; \
 	MONITOR_DEF monitors[16]; \
 	MONITOR_DEF virtualScreen; \
-	HANDLE updateEvent; \
+	rdpShadowMultiClientEvent* updateEvent; \
 	BOOL suppressOutput; \
 	REGION16 invalidRegion; \
 	wMessagePipe* MsgPipe; \
-	SYNCHRONIZATION_BARRIER barrier; \
 	UINT32 pointerX; \
 	UINT32 pointerY; \
 	\

server/shadow/CMakeLists.txt

@@ -171,6 +171,8 @@ set(${MODULE_PREFIX}_SRCS
 	shadow_remdesk.h
 	shadow_subsystem.c
 	shadow_subsystem.h
+	shadow_mcevent.c
+	shadow_mcevent.h
 	shadow_server.c
 	shadow.h)
 

server/shadow/Mac/mac_shadow.c

@@ -31,6 +31,7 @@
 #include "../shadow_capture.h"
 #include "../shadow_encoder.h"
 #include "../shadow_subsystem.h"
+#include "../shadow_mcevent.h"
 
 #include "mac_shadow.h"
 
@@ -366,13 +367,7 @@ void (^mac_capture_stream_handler)(CGDisplayStreamFrameStatus, uint64_t, IOSurfa
 			
 		count = ArrayList_Count(server->clients);
 			
-		InitializeSynchronizationBarrier(&(subsystem->barrier), count + 1, -1);
+		shadow_multiclient_publish_and_wait(subsystem->updateEvent);
-			
-		SetEvent(subsystem->updateEvent);
-			
-		EnterSynchronizationBarrier(&(subsystem->barrier), 0);
-		
-		DeleteSynchronizationBarrier(&(subsystem->barrier));
 		
 		if (count == 1)
 		{
@@ -386,8 +381,6 @@ void (^mac_capture_stream_handler)(CGDisplayStreamFrameStatus, uint64_t, IOSurfa
 			}
 		}
 		
-		ResetEvent(subsystem->updateEvent);
-			
 		ArrayList_Unlock(server->clients);
 			
 		region16_clear(&(subsystem->invalidRegion));

server/shadow/Win/win_shadow.c

@@ -28,6 +28,7 @@
 #include "../shadow_surface.h"
 #include "../shadow_capture.h"
 #include "../shadow_subsystem.h"
+#include "../shadow_mcevent.h"
 
 #include "win_shadow.h"
 
@@ -287,14 +288,7 @@ int win_shadow_surface_copy(winShadowSubsystem* subsystem)
 
 	count = ArrayList_Count(server->clients);
 
-	InitializeSynchronizationBarrier(&(subsystem->barrier), count + 1, -1);
+	shadow_multiclient_publish_and_wait(subsystem->updateEvent);
-
-	SetEvent(subsystem->updateEvent);
-
-	EnterSynchronizationBarrier(&(subsystem->barrier), 0);
-	ResetEvent(subsystem->updateEvent);
-
-	DeleteSynchronizationBarrier(&(subsystem->barrier));
 
 	ArrayList_Unlock(server->clients);
 

server/shadow/X11/x11_shadow.c

@@ -46,6 +46,7 @@
 #include "../shadow_capture.h"
 #include "../shadow_surface.h"
 #include "../shadow_subsystem.h"
+#include "../shadow_mcevent.h"
 
 #include "x11_shadow.h"
 
@@ -708,13 +709,7 @@ int x11_shadow_screen_grab(x11ShadowSubsystem* subsystem)
 
 		count = ArrayList_Count(server->clients);
 
-		InitializeSynchronizationBarrier(&(subsystem->barrier), count + 1, -1);
+		shadow_multiclient_publish_and_wait(subsystem->updateEvent);
-
-		SetEvent(subsystem->updateEvent);
-
-		EnterSynchronizationBarrier(&(subsystem->barrier), 0);
-
-		DeleteSynchronizationBarrier(&(subsystem->barrier));
 
 		if (count == 1)
 		{
@@ -728,8 +723,6 @@ int x11_shadow_screen_grab(x11ShadowSubsystem* subsystem)
 			}
 		}
 
-		ResetEvent(subsystem->updateEvent);
-
 		region16_clear(&(subsystem->invalidRegion));
 	}
 

server/shadow/shadow.h

@@ -30,6 +30,7 @@
 #include "shadow_channels.h"
 #include "shadow_subsystem.h"
 #include "shadow_lobby.h"
+#include "shadow_mcevent.h"
 
 #ifdef __cplusplus
 extern "C" {

server/shadow/shadow_client.c

@@ -949,6 +949,7 @@ void* shadow_client_thread(rdpShadowClient* client)
 	HANDLE StopEvent;
 	HANDLE ClientEvent;
 	HANDLE ChannelEvent;
+	void* UpdateSubscriber;
 	HANDLE UpdateEvent;
 	freerdp_peer* peer;
 	rdpContext* context;
@@ -980,8 +981,15 @@ void* shadow_client_thread(rdpShadowClient* client)
 	peer->update->SuppressOutput = (pSuppressOutput) shadow_client_suppress_output;
 	peer->update->SurfaceFrameAcknowledge = (pSurfaceFrameAcknowledge) shadow_client_surface_frame_acknowledge;
 
+	if ((!client->StopEvent) || (!client->vcm) || (!subsystem->updateEvent))
+		goto out;
+
+	UpdateSubscriber = shadow_multiclient_get_subscriber(subsystem->updateEvent);
+	if (!UpdateSubscriber)
+		goto out;
+
 	StopEvent = client->StopEvent;
-	UpdateEvent = subsystem->updateEvent;
+	UpdateEvent = shadow_multiclient_getevent(UpdateSubscriber);
 	ClientEvent = peer->GetEventHandle(peer);
 	ChannelEvent = WTSVirtualChannelManagerGetEventHandle(client->vcm);
 
@@ -998,11 +1006,6 @@ void* shadow_client_thread(rdpShadowClient* client)
 
 		if (WaitForSingleObject(StopEvent, 0) == WAIT_OBJECT_0)
 		{
-			if (WaitForSingleObject(UpdateEvent, 0) == WAIT_OBJECT_0)
-			{
-				EnterSynchronizationBarrier(&(subsystem->barrier), 0);
-			}
-
 			break;
 		}
 
@@ -1024,9 +1027,11 @@ void* shadow_client_thread(rdpShadowClient* client)
 				shadow_client_send_surface_update(client);
 			}
 
-			EnterSynchronizationBarrier(&(subsystem->barrier), 0);
+			/* 
-
+			 * The return value of shadow_multiclient_consume is whether or not the subscriber really consumes the event.
-			while (WaitForSingleObject(UpdateEvent, 0) == WAIT_OBJECT_0);
+			 * It's not cared currently.
+			 */
+			(void)shadow_multiclient_consume(UpdateSubscriber);
 		}
 
 		if (WaitForSingleObject(ClientEvent, 0) == WAIT_OBJECT_0)
@@ -1059,6 +1064,13 @@ void* shadow_client_thread(rdpShadowClient* client)
 		}
 	}
 
+	if (UpdateSubscriber)
+	{
+		shadow_multiclient_release_subscriber(UpdateSubscriber);
+		UpdateSubscriber = NULL;
+	}
+
+out:
 	peer->Disconnect(peer);
 	
 	freerdp_peer_context_free(peer);

server/shadow/shadow_mcevent.c

@@ -0,0 +1,337 @@
+/**
+ * FreeRDP: A Remote Desktop Protocol Implementation
+ *
+ * Copyright 2015 Jiang Zihao <zihao.jiang@yahoo.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <assert.h>
+#include <freerdp/log.h>
+#include "shadow.h"
+
+#define TAG SERVER_TAG("shadow.mcevent")
+
+struct rdp_shadow_multiclient_subscriber
+{
+	rdpShadowMultiClientEvent* ref;
+	BOOL pleaseHandle; /* Indicate if server expects my handling in this turn */
+};
+
+rdpShadowMultiClientEvent* shadow_multiclient_new()
+{
+	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;
+	SetEvent(event->doneEvent);
+	return event;
+
+out_free_subscribers:
+	ArrayList_Free(event->subscribers);
+out_free_doneEvent:
+	CloseHandle(event->doneEvent);
+out_free_barrierEvent:
+	CloseHandle(event->barrierEvent);
+out_free_event:
+	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);
+	CloseHandle(event->doneEvent);
+	CloseHandle(event->barrierEvent);
+	CloseHandle(event->event);
+	free(event);
+
+	return;
+}
+
+static void _Publish(rdpShadowMultiClientEvent* event)
+{
+	wArrayList* subscribers;
+	struct rdp_shadow_multiclient_subscriber* subscriber = NULL;
+	int i;
+
+	subscribers = event->subscribers;
+
+	assert(event->consuming == 0);
+
+	/* Count subscribing clients */
+	ArrayList_Lock(subscribers);
+	for (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);
+		ResetEvent(event->doneEvent);
+		SetEvent(event->event);
+	}
+
+	return;
+}
+
+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));
+		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 */
+	assert(WaitForSingleObject(event->event, 0) != WAIT_OBJECT_0);
+
+	return;
+}
+
+void shadow_multiclient_publish(rdpShadowMultiClientEvent* event)
+{
+	if (!event)
+		return;
+
+	EnterCriticalSection(&(event->lock));
+	_Publish(event);
+	LeaveCriticalSection(&(event->lock));
+
+	return;
+}
+void shadow_multiclient_wait(rdpShadowMultiClientEvent* event)
+{
+	if (!event)
+		return;
+
+	EnterCriticalSection(&(event->lock));
+	_WaitForSubscribers(event);
+	LeaveCriticalSection(&(event->lock));
+
+	return;
+}
+void shadow_multiclient_publish_and_wait(rdpShadowMultiClientEvent* event)
+{
+	if (!event)
+		return;
+
+	EnterCriticalSection(&(event->lock));
+	_Publish(event);
+	_WaitForSubscribers(event);
+	LeaveCriticalSection(&(event->lock));
+
+	return;
+}
+
+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;
+	}
+
+	assert(event->consuming >= 0);
+
+	if (event->consuming == 0)
+	{
+		/* Last client reset event before notify clients to continue */
+		ResetEvent(event->event);
+
+		if (event->waiting > 0)
+		{
+			/* Notify other clients to continue */
+			SetEvent(event->barrierEvent);
+		}
+		else
+		{
+			/* Only one client. Notify server directly */
+			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));
+			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.
+				 */
+				ResetEvent(event->barrierEvent);
+				SetEvent(event->doneEvent);
+			}
+		}
+	}
+
+	return ret;
+}
+
+void* shadow_multiclient_get_subscriber(rdpShadowMultiClientEvent* event)
+{
+	struct rdp_shadow_multiclient_subscriber* subscriber;
+
+	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_Add(event->subscribers, subscriber) < 0)
+		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;
+	rdpShadowMultiClientEvent* event;
+
+	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);
+
+	return;
+}
+
+BOOL shadow_multiclient_consume(void* subscriber)
+{
+	struct rdp_shadow_multiclient_subscriber* s;
+	rdpShadowMultiClientEvent* event;
+	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;
+}

server/shadow/shadow_mcevent.h

@@ -0,0 +1,65 @@
+/**
+ * FreeRDP: A Remote Desktop Protocol Implementation
+ *
+ * Copyright 2015 Jiang Zihao <zihao.jiang@yahoo.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef FREERDP_SHADOW_SERVER_MCEVENT_H
+#define FREERDP_SHADOW_SERVER_MCEVENT_H
+
+#include <freerdp/server/shadow.h>
+
+#include <winpr/crt.h>
+#include <winpr/synch.h>
+#include <winpr/collections.h>
+
+/*
+ * This file implemented a model that an event is consumed 
+ * by multiple clients. All clients should wait others before continue
+ * Server should wait for all clients before continue
+ */
+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;
+};
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+rdpShadowMultiClientEvent* shadow_multiclient_new();
+void shadow_multiclient_free(rdpShadowMultiClientEvent* event);
+void shadow_multiclient_publish(rdpShadowMultiClientEvent* event);
+void shadow_multiclient_wait(rdpShadowMultiClientEvent* event);
+void shadow_multiclient_publish_and_wait(rdpShadowMultiClientEvent* event);
+void* shadow_multiclient_get_subscriber(rdpShadowMultiClientEvent* event);
+void shadow_multiclient_release_subscriber(void* subscriber);
+BOOL shadow_multiclient_consume(void* subscriber);
+HANDLE shadow_multiclient_getevent(void* subscriber);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* FREERDP_SHADOW_SERVER_MCEVENT_H */

server/shadow/shadow_subsystem.c

@@ -143,7 +143,7 @@ int shadow_subsystem_init(rdpShadowSubsystem* subsystem, rdpShadowServer* server
 	if (!(subsystem->MsgPipe = MessagePipe_New()))
 		goto fail;
 
-	if (!(subsystem->updateEvent = CreateEvent(NULL, TRUE, FALSE, NULL)))
+	if (!(subsystem->updateEvent = shadow_multiclient_new()))
 		goto fail;
 
 	region16_init(&(subsystem->invalidRegion));
@@ -160,7 +160,7 @@ fail:
 
 	if (subsystem->updateEvent)
 	{
-		CloseHandle(subsystem->updateEvent);
+		shadow_multiclient_free(subsystem->updateEvent);
 		subsystem->updateEvent = NULL;
 	}
 
@@ -183,7 +183,7 @@ void shadow_subsystem_uninit(rdpShadowSubsystem* subsystem)
 
 	if (subsystem->updateEvent)
 	{
-		CloseHandle(subsystem->updateEvent);
+		shadow_multiclient_free(subsystem->updateEvent);
 		subsystem->updateEvent = NULL;
 	}