+++ /dev/null
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- */
-
-using System;
-using System.Collections.Generic;
-using System.Threading;
-using Thrift.Collections;
-using Thrift.Protocol;
-using Thrift.Transport;
-
-namespace Thrift.Server
-{
- /// <summary>
- /// Server that uses C# threads (as opposed to the ThreadPool) when handling requests.
- /// </summary>
- public class TThreadedServer : TServer
- {
- private const int DEFAULT_MAX_THREADS = 100;
- private volatile bool stop = false;
- private readonly int maxThreads;
-
- private Queue<TTransport> clientQueue;
- private THashSet<Thread> clientThreads;
- private object clientLock;
- private Thread workerThread;
-
- public int ClientThreadsCount
- {
- get { return clientThreads.Count; }
- }
-
- public TThreadedServer(TProcessor processor, TServerTransport serverTransport)
- : this(new TSingletonProcessorFactory(processor), serverTransport,
- new TTransportFactory(), new TTransportFactory(),
- new TBinaryProtocol.Factory(), new TBinaryProtocol.Factory(),
- DEFAULT_MAX_THREADS, DefaultLogDelegate)
- {
- }
-
- public TThreadedServer(TProcessor processor, TServerTransport serverTransport, LogDelegate logDelegate)
- : this(new TSingletonProcessorFactory(processor), serverTransport,
- new TTransportFactory(), new TTransportFactory(),
- new TBinaryProtocol.Factory(), new TBinaryProtocol.Factory(),
- DEFAULT_MAX_THREADS, logDelegate)
- {
- }
-
-
- public TThreadedServer(TProcessor processor,
- TServerTransport serverTransport,
- TTransportFactory transportFactory,
- TProtocolFactory protocolFactory)
- : this(new TSingletonProcessorFactory(processor), serverTransport,
- transportFactory, transportFactory,
- protocolFactory, protocolFactory,
- DEFAULT_MAX_THREADS, DefaultLogDelegate)
- {
- }
-
- public TThreadedServer(TProcessorFactory processorFactory,
- TServerTransport serverTransport,
- TTransportFactory transportFactory,
- TProtocolFactory protocolFactory)
- : this(processorFactory, serverTransport,
- transportFactory, transportFactory,
- protocolFactory, protocolFactory,
- DEFAULT_MAX_THREADS, DefaultLogDelegate)
- {
- }
- public TThreadedServer(TProcessorFactory processorFactory,
- TServerTransport serverTransport,
- TTransportFactory inputTransportFactory,
- TTransportFactory outputTransportFactory,
- TProtocolFactory inputProtocolFactory,
- TProtocolFactory outputProtocolFactory,
- int maxThreads, LogDelegate logDel)
- : base(processorFactory, serverTransport, inputTransportFactory, outputTransportFactory,
- inputProtocolFactory, outputProtocolFactory, logDel)
- {
- this.maxThreads = maxThreads;
- clientQueue = new Queue<TTransport>();
- clientLock = new object();
- clientThreads = new THashSet<Thread>();
- }
-
- /// <summary>
- /// Use new Thread for each new client connection. block until numConnections < maxThreads.
- /// </summary>
- public override void Serve()
- {
- try
- {
- //start worker thread
- workerThread = new Thread(new ThreadStart(Execute));
- workerThread.Start();
- serverTransport.Listen();
- }
- catch (TTransportException ttx)
- {
- logDelegate("Error, could not listen on ServerTransport: " + ttx);
- return;
- }
-
- //Fire the preServe server event when server is up but before any client connections
- if (serverEventHandler != null)
- serverEventHandler.preServe();
-
- while (!stop)
- {
- int failureCount = 0;
- try
- {
- TTransport client = serverTransport.Accept();
- lock (clientLock)
- {
- clientQueue.Enqueue(client);
- Monitor.Pulse(clientLock);
- }
- }
- catch (TTransportException ttx)
- {
- if (!stop || ttx.Type != TTransportException.ExceptionType.Interrupted)
- {
- ++failureCount;
- logDelegate(ttx.ToString());
- }
-
- }
- }
-
- if (stop)
- {
- try
- {
- serverTransport.Close();
- }
- catch (TTransportException ttx)
- {
- logDelegate("TServeTransport failed on close: " + ttx.Message);
- }
- stop = false;
- }
- }
-
- /// <summary>
- /// Loops on processing a client forever
- /// </summary>
- private void Execute()
- {
- while (!stop)
- {
- TTransport client;
- Thread t;
- lock (clientLock)
- {
- //don't dequeue if too many connections
- while (clientThreads.Count >= maxThreads)
- {
- Monitor.Wait(clientLock);
- }
-
- while (clientQueue.Count == 0)
- {
- Monitor.Wait(clientLock);
- }
-
- client = clientQueue.Dequeue();
- t = new Thread(new ParameterizedThreadStart(ClientWorker));
- clientThreads.Add(t);
- }
- //start processing requests from client on new thread
- t.Start(client);
- }
- }
-
- private void ClientWorker(object context)
- {
- using (TTransport client = (TTransport)context)
- {
- TProcessor processor = processorFactory.GetProcessor(client);
- TTransport inputTransport = null;
- TTransport outputTransport = null;
- TProtocol inputProtocol = null;
- TProtocol outputProtocol = null;
- object connectionContext = null;
- try
- {
- try
- {
- inputTransport = inputTransportFactory.GetTransport(client);
- outputTransport = outputTransportFactory.GetTransport(client);
- inputProtocol = inputProtocolFactory.GetProtocol(inputTransport);
- outputProtocol = outputProtocolFactory.GetProtocol(outputTransport);
-
- //Recover event handler (if any) and fire createContext server event when a client connects
- if (serverEventHandler != null)
- connectionContext = serverEventHandler.createContext(inputProtocol, outputProtocol);
-
- //Process client requests until client disconnects
- while (!stop)
- {
- if (!inputTransport.Peek())
- break;
-
- //Fire processContext server event
- //N.B. This is the pattern implemented in C++ and the event fires provisionally.
- //That is to say it may be many minutes between the event firing and the client request
- //actually arriving or the client may hang up without ever makeing a request.
- if (serverEventHandler != null)
- serverEventHandler.processContext(connectionContext, inputTransport);
- //Process client request (blocks until transport is readable)
- if (!processor.Process(inputProtocol, outputProtocol))
- break;
- }
- }
- catch (TTransportException)
- {
- //Usually a client disconnect, expected
- }
- catch (Exception x)
- {
- //Unexpected
- logDelegate("Error: " + x);
- }
-
- //Fire deleteContext server event after client disconnects
- if (serverEventHandler != null)
- serverEventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
-
- lock (clientLock)
- {
- clientThreads.Remove(Thread.CurrentThread);
- Monitor.Pulse(clientLock);
- }
-
- }
- finally
- {
- //Close transports
- if (inputTransport != null)
- inputTransport.Close();
- if (outputTransport != null)
- outputTransport.Close();
-
- // disposable stuff should be disposed
- if (inputProtocol != null)
- inputProtocol.Dispose();
- if (outputProtocol != null)
- outputProtocol.Dispose();
- }
- }
- }
-
- public override void Stop()
- {
- stop = true;
- serverTransport.Close();
- //clean up all the threads myself
- workerThread.Abort();
- foreach (Thread t in clientThreads)
- {
- t.Abort();
- }
- }
- }
-}
projects / ceph.git / blobdiff