scsman.cxx

Windows CE 6.0 Server 源码

//
// Copyright (c) Microsoft Corporation.  All rights reserved.
//
//
// Use of this source code is subject to the terms of the Microsoft shared
// source or premium shared source license agreement under which you licensed
// this source code. If you did not accept the terms of the license agreement,
// you are not authorized to use this source code. For the terms of the license,
// please see the license agreement between you and Microsoft or, if applicable,
// see the SOURCE.RTF on your install media or the root of your tools installation.
// THE SOURCE CODE IS PROVIDED "AS IS", WITH NO WARRANTIES.
//
/*++


Module Name:

    scsman.cxx

Abstract:

    Small client session manager

--*/

#include <sc.hxx>
#include <scqman.hxx>
#include <scqueue.hxx>
#include <scutil.hxx>
#include <scsman.hxx>
#include <scpacket.hxx>
#include <scnb.hxx>

#include <scping.hxx>

#include <sccomp.hxx>

#include <ph.h>
#include <phintr.h>
#include <_mqini.h>

#define SC_THREAD_WAIT				 5000		// 5 seconds
#define SC_THREAD_WAIT2				 500

HANDLE ScSessionManager::hNetUP = NULL;

ScSessionManager::ScSessionManager (void) {
	SVSUTIL_ASSERT (gMem->IsLocked());

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_INIT, L"Creating Session Manager...\n");
#endif

	fBusy = 0;

	fInitialized = FALSE;
	pSessList    = NULL;
	s_listen     = INVALID_SOCKET;
	iThreadPairs = 0;

	hAccThread = NULL;

	hBuzz = CreateEvent (NULL, FALSE, FALSE, NULL);
	if (! hBuzz)
		return;

	if (gMachine->fNetworkTracking) {
		hNetUP = CreateEvent (NULL, FALSE, FALSE, NULL);
		if (! hNetUP)
			return;
	}

	if (! gMachine->fUseBinary) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Binary protocol disabled - NOT creating binary ports...\n");
#endif
		fInitialized = TRUE;
		return;
	}

	s_listen = socket (AF_INET, SOCK_STREAM, 0);

	if (s_listen == INVALID_SOCKET)
		return;

	sockaddr_in sa;
	memset (&sa, 0, sizeof(sa));
	sa.sin_family = AF_INET;
	sa.sin_port   = htons (gMachine->uiPort);
	sa.sin_addr.S_un.S_addr = htonl (INADDR_ANY);

	if (bind (s_listen, (sockaddr *)&sa, sizeof(sa))) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"port binding error: %08x\n", WSAGetLastError ());
#endif
 		return;
	}

	if (listen (s_listen, 5)) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"listen error: %08x\n", WSAGetLastError ());
#endif
		return;
	}

	fInitialized = TRUE;
}

void ScSessionManager::Start(void) {
	SVSUTIL_ASSERT (gMem->IsLocked());

	if (! gMachine->fUseBinary) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Binary protocol disabled - NOT starting binary listener...\n");
#endif
		return;
	}

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION, L"Starting listener thread...\n");
#endif

	StartPingServer ();

	DWORD tid;
	hAccThread  = CreateThread (NULL, 0, ScSessionManager::AccThread_s,  NULL, 0, &tid);
}

void ScSessionManager::Stop(void) {
	SVSUTIL_ASSERT (gMem->IsLocked());

	if (gMachine->fUseBinary)
		StopPingServer ();

	//
	//	First, close all sockets
	//
#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION, L"Wrapping up communications...\n");
#endif

	if (s_listen != INVALID_SOCKET) {
		closesocket (s_listen);
		s_listen = INVALID_SOCKET;
	}

	ScSession *pList = pSessList;
	while (pList) {
		pList->fSessionState = SCSESSION_STATE_EXITING;		// Signal reader
		SetEvent (pList->hEvent);							// ...and writer

		if (pList->s != INVALID_SOCKET) {
			closesocket (pList->s);
			pList->s = INVALID_SOCKET;
			pList->ipPeerAddr.S_un.S_addr = INADDR_NONE;
		}

		// if we're using wininet, close connection from under it.
		if (pList->hInternetSession)
			pList->SRMPCloseSession();

		pList = pList->pNext;
	}

	//
	//	Now kill all threads
	//
#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION, L"Stopping listener thread...\n");
#endif
	if (hAccThread) {
		if (WaitForSingleObject (hAccThread, SC_THREAD_WAIT) == WAIT_TIMEOUT) {
			scerror_Inform (MSMQ_SC_ERRMSG_LISTENKILLED);
			TerminateThread (hAccThread, 0);
		}

		CloseHandle (hAccThread);
	}

	hAccThread = NULL;

	//
	//	Stop all session threads...
	//
	pList = pSessList;
	while (pList) {
		if (pList->hServiceThreadR) {
#if defined (SC_VERBOSE)
		    scerror_DebugOut (VERBOSE_MASK_SESSION, L"Terminating receiver for %s...\n", pList->lpszHostName);
#endif
			if (WaitForSingleObject (pList->hServiceThreadR, SC_THREAD_WAIT) == WAIT_TIMEOUT) {
				scerror_Inform (MSMQ_SC_ERRMSG_READERKILLED, pList->lpszHostName);
				TerminateThread (pList->hServiceThreadR, 0);
			}

			CloseHandle (pList->hServiceThreadR);
		}
		pList->hServiceThreadR = NULL;

		if (pList->hServiceThreadW) {
#if defined (SC_VERBOSE)
		    scerror_DebugOut (VERBOSE_MASK_SESSION, L"Terminating sender for %s...\n", pList->lpszHostName);
#endif
			if (WaitForSingleObject (pList->hServiceThreadW, SC_THREAD_WAIT) == WAIT_TIMEOUT) {
				scerror_Inform (MSMQ_SC_ERRMSG_WRITERKILLED, pList->lpszHostName);
				TerminateThread (pList->hServiceThreadW, 0);
			}

			CloseHandle (pList->hServiceThreadW);
		}
		pList->hServiceThreadW = NULL;
		pList = pList->pNext;
	}
}

ScSessionManager::~ScSessionManager (void) {
	SVSUTIL_ASSERT (gMem->IsLocked());

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_INIT, L"Freeing Session Manager...\n");
#endif

	SVSUTIL_ASSERT (! hAccThread);

	if (hBuzz)
		CloseHandle (hBuzz);

	if (hNetUP)
		CloseHandle (hNetUP);

	hNetUP = NULL;

	ScSession *pList = pSessList;

	while (pList) {
		ScSession *pNext = pList->pNext;

		SVSUTIL_ASSERT (! pList->hServiceThreadW);
		SVSUTIL_ASSERT (! pList->hServiceThreadR);

		if (pList->lpszHostName) {
			svsutil_StringHashFree (gMem->pStringHash, pList->lpszHostName);
			pList->lpszHostName = NULL;
		}

		delete pList;

		pList = pNext;
	}

	if (s_listen != INVALID_SOCKET)
		closesocket (s_listen);
}

ScSession *ScSessionManager::GetSession (WCHAR *a_lpszHostName, int qType) {
	SVSUTIL_ASSERT (gMem->IsLocked());

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION, L"Acquiring session for session for %s... ", a_lpszHostName);
#endif

	svsutil_StringHashRef (gMem->pStringHash, a_lpszHostName);

	ScSession *pSess = pSessList;
	while (pSess) {
		if (pSess->lpszHostName == a_lpszHostName && qType == pSess->qType) {
			pSess->AddRef ();
			if (a_lpszHostName)
				svsutil_StringHashFree (gMem->pStringHash, a_lpszHostName);		// compensate the ref

#if defined (SC_VERBOSE)
			scerror_DebugOut (VERBOSE_MASK_SESSION, L"Failed...\n");
#endif
			return pSess;
		}
		pSess = pSess->pNext;
	}

	pSess = new ScSession (a_lpszHostName, pSessList, qType);
	if (pSess) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Created...\n");
#endif
		pSessList = pSess;
	} else {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Failed...\n");
#endif
		SVSUTIL_ASSERT (0);

		if (a_lpszHostName)
			svsutil_StringHashFree (gMem->pStringHash, a_lpszHostName);		// compensate the ref
	}

	return pSess;
}

ScSession *ScSessionManager::GetInactiveOsSession (WCHAR *a_lpszHostName) {
	SVSUTIL_ASSERT (gMem->IsLocked());

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION, L"Creating new (duplicate check override) session for session for %s... ", a_lpszHostName);
#endif

	svsutil_StringHashRef (gMem->pStringHash, a_lpszHostName);

	ScSession *pSess = pSessList;
	while (pSess) {
		if ((pSess->lpszHostName == a_lpszHostName) &&
		    ((pSess->qType == SCFILE_QP_FORMAT_TCP) ||
		     (pSess->qType == SCFILE_QP_FORMAT_OS)) && 
			((pSess->fSessionState == SCSESSION_STATE_INACTIVE) ||
			 (pSess->fSessionState == SCSESSION_STATE_WAITING))) {

			pSess->AddRef ();
			if (a_lpszHostName)
				svsutil_StringHashFree (gMem->pStringHash, a_lpszHostName);		// compensate the ref

#if defined (SC_VERBOSE)
			scerror_DebugOut (VERBOSE_MASK_SESSION, L"Failed...\n");
#endif
			return pSess;
		}
		pSess = pSess->pNext;
	}

	pSess = new ScSession (a_lpszHostName, pSessList, SCFILE_QP_FORMAT_OS);

	if (pSess) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Created...\n");
#endif
		pSessList = pSess;
	} else {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Failed...\n");
#endif
		SVSUTIL_ASSERT (0);

		if (a_lpszHostName)
			svsutil_StringHashFree (gMem->pStringHash, a_lpszHostName);		// compensate the ref
	}

	return pSess;
}

void ScSessionManager::ReleaseSession (ScSession *pSess) {
	SVSUTIL_ASSERT (gMem->IsLocked());

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION, L"Releasing session for %s\n", pSess->lpszHostName);
#endif
	pSess->DelRef ();
	if (pSess->GetRefCount() == 0) {
		if (pSess->pPrev == NULL) {
			SVSUTIL_ASSERT (pSess == pSessList);
			pSessList = pSessList->pNext;
			if (pSessList)
				pSessList->pPrev = NULL;
		} else {
			pSess->pPrev->pNext = pSess->pNext;

			if (pSess->pNext)
				pSess->pNext->pPrev = pSess->pPrev;
		}
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Permanently deleting session for %s\n", pSess->lpszHostName);
#endif
		if (pSess->lpszHostName) {
			svsutil_StringHashFree (gMem->pStringHash, pSess->lpszHostName);
			pSess->lpszHostName = NULL;
		}
		delete pSess;

		return;
	}
}

void ScSessionManager::PacketInserted (ScSession *pSess) {
	SVSUTIL_ASSERT (gMem->IsLocked());

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION | VERBOSE_MASK_PACKETS | VERBOSE_MASK_QUEUE, L"Bzzzzz!!! Packet Inserted...\n");
#endif

	if ((! pSess) || (pSess->fSessionState == SCSESSION_STATE_INACTIVE))
		SetEvent (hBuzz);
	else if (pSess->fSessionState == SCSESSION_STATE_OPERATING)
		SetEvent (pSess->hEvent);
#if defined (SC_VERBOSE)
	else
		scerror_DebugOut (VERBOSE_MASK_SESSION | VERBOSE_MASK_PACKETS | VERBOSE_MASK_QUEUE, L"...silent buzz, actually...\n");
#endif
}

//
//	private support
//
int ScSessionManager::SpinSession (ScSession *pSess, SOCKET s) {
	SVSUTIL_ASSERT (gMem->IsLocked());

	//
	//	If session already exists and active, we don't want to duplicate the
	//	activity. Let them just send through already existing connection!
	//
	if ((pSess->fSessionState != SCSESSION_STATE_INACTIVE) &&
		(pSess->fSessionState != SCSESSION_STATE_WAITING))
		return FALSE;

#if defined (SC_VERBOSE)
	scerror_DebugOut (VERBOSE_MASK_SESSION, L"Preparing to spin thread pair for %s...\n", pSess->lpszHostName);
#endif

	//
	//	We need thread pair to spin a connection. Do we have it?
	//
	if (! BookThreadPair()) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"No free thread pairs...\n");
#endif
		return FALSE;
	}

	pSess->s			 = s;
	pSess->fSessionState = SCSESSION_STATE_CONNECTING;
	pSess->hEvent        = CreateEvent (NULL, FALSE, FALSE, NULL);

	DWORD tid;

	if (pSess->qType == SCFILE_QP_FORMAT_HTTP || pSess->qType == SCFILE_QP_FORMAT_HTTPS) {
		pSess->hServiceThreadW = CreateThread (NULL, 0, ScSession::ServiceThreadHttpW_s,  pSess, 0, &tid);
		return TRUE;
	}
	
	if (! gMachine->fUseBinary) {
#if defined (SC_VERBOSE)
		scerror_DebugOut (VERBOSE_MASK_SESSION, L"Binary protocol disabled - session order ignored...\n");
#endif
		ReleaseThreadPair();
		return FALSE;
	}