From e8438b808460ed2adaead034ee727f22671f157c Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 8 Oct 2012 11:04:11 -0500 Subject: [PATCH] clean up Thrift session management and encapsulate in ThriftSessionManager patch by jbellis; reviewed by dbrosius for CASSANDRA-4657 --- .../apache/cassandra/service/ClientState.java | 24 ++------ .../SocketSessionManagementService.java | 59 ------------------- .../cassandra/thrift/CassandraServer.java | 31 ++-------- .../cassandra/thrift/CustomTHsHaServer.java | 9 +-- .../thrift/CustomTNonBlockingServer.java | 4 +- .../thrift/CustomTThreadPoolServer.java | 42 ++++--------- .../TCustomNonblockingServerSocket.java | 9 +-- 7 files changed, 36 insertions(+), 142 deletions(-) delete mode 100644 src/java/org/apache/cassandra/service/SocketSessionManagementService.java diff --git a/src/java/org/apache/cassandra/service/ClientState.java b/src/java/org/apache/cassandra/service/ClientState.java index 35e382863e..c95e3fb377 100644 --- a/src/java/org/apache/cassandra/service/ClientState.java +++ b/src/java/org/apache/cassandra/service/ClientState.java @@ -77,12 +77,15 @@ public class ClientState } /** - * Construct a new, empty ClientState: can be reused after logout() or reset(). + * Construct a new, empty ClientState */ public ClientState(boolean internalCall) { this.internalCall = internalCall; - reset(); + + user = DatabaseDescriptor.getAuthenticator().defaultUser(); + resourceClear(); + prepared.clear(); } public Map getPrepared() @@ -159,13 +162,6 @@ public class ClientState this.user = user; } - public void logout() - { - if (logger.isDebugEnabled()) - logger.debug("logged out: {}", user); - reset(); - } - private void resourceClear() { resource.clear(); @@ -173,16 +169,6 @@ public class ClientState resource.add(Resources.KEYSPACES); } - public void reset() - { - user = DatabaseDescriptor.getAuthenticator().defaultUser(); - keyspace = null; - preparedTracingSession = null; - resourceClear(); - prepared.clear(); - cqlVersion = DEFAULT_CQL_VERSION; - } - public void hasKeyspaceAccess(String keyspace, Permission perm) throws UnauthorizedException, InvalidRequestException { hasColumnFamilySchemaAccess(keyspace, perm); diff --git a/src/java/org/apache/cassandra/service/SocketSessionManagementService.java b/src/java/org/apache/cassandra/service/SocketSessionManagementService.java deleted file mode 100644 index 423d454015..0000000000 --- a/src/java/org/apache/cassandra/service/SocketSessionManagementService.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * 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. - */ -package org.apache.cassandra.service; - -import java.net.SocketAddress; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -public class SocketSessionManagementService -{ - public final static SocketSessionManagementService instance = new SocketSessionManagementService(); - public final static ThreadLocal remoteSocket = new ThreadLocal(); - private final Map activeSocketSessions = new ConcurrentHashMap(); - - public ClientState get(SocketAddress key) - { - ClientState retval = null; - if (key != null) - { - retval = activeSocketSessions.get(key); - } - return retval; - } - - public void put(SocketAddress key, ClientState value) - { - if (key != null && value != null) - { - activeSocketSessions.put(key, value); - } - } - - public boolean remove(SocketAddress key) - { - assert key != null; - return activeSocketSessions.remove(key) != null; - } - - public void clear() - { - activeSocketSessions.clear(); - } - -} diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 4ab19bb846..6a5ef45d4e 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -19,7 +19,6 @@ package org.apache.cassandra.thrift; import java.io.IOException; import java.io.UnsupportedEncodingException; -import java.net.SocketAddress; import java.nio.ByteBuffer; import java.nio.charset.CharacterCodingException; import java.util.*; @@ -35,8 +34,10 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.auth.Permission; -import org.apache.cassandra.auth.PermissionDenied; -import org.apache.cassandra.config.*; +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.KSMetaData; +import org.apache.cassandra.config.Schema; import org.apache.cassandra.cql.CQLStatement; import org.apache.cassandra.cql.QueryProcessor; import org.apache.cassandra.db.*; @@ -46,9 +47,9 @@ import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.db.marshal.TimeUUIDType; import org.apache.cassandra.dht.*; +import org.apache.cassandra.exceptions.ReadTimeoutException; import org.apache.cassandra.exceptions.RequestExecutionException; import org.apache.cassandra.exceptions.RequestValidationException; -import org.apache.cassandra.exceptions.ReadTimeoutException; import org.apache.cassandra.exceptions.UnauthorizedException; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.locator.DynamicEndpointSnitch; @@ -70,16 +71,6 @@ public class CassandraServer implements Cassandra.Iface private final static List EMPTY_SUBCOLUMNS = Collections.emptyList(); private final static List EMPTY_COUNTER_SUBCOLUMNS = Collections.emptyList(); - // thread local state containing session information - public final ThreadLocal clientState = new ThreadLocal() - { - @Override - public ClientState initialValue() - { - return new ClientState(); - } - }; - /* * RequestScheduler to perform the scheduling of incoming requests */ @@ -92,17 +83,7 @@ public class CassandraServer implements Cassandra.Iface public ClientState state() { - SocketAddress remoteSocket = SocketSessionManagementService.remoteSocket.get(); - if (remoteSocket == null) - return clientState.get(); - - ClientState cState = SocketSessionManagementService.instance.get(remoteSocket); - if (cState == null) - { - cState = new ClientState(); - SocketSessionManagementService.instance.put(remoteSocket, cState); - } - return cState; + return ThriftSessionManager.instance.currentSession(); } protected Map readColumnFamily(List commands, org.apache.cassandra.db.ConsistencyLevel consistency_level) diff --git a/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java b/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java index 436dbb3514..4d3aa62ff7 100644 --- a/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java +++ b/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java @@ -30,18 +30,19 @@ import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.service.SocketSessionManagementService; +import org.apache.cassandra.service.ThriftSessionManager; import org.apache.thrift.server.TNonblockingServer; import org.apache.thrift.server.TServer; import org.apache.thrift.transport.TNonblockingServerTransport; import org.apache.thrift.transport.TNonblockingSocket; import org.apache.thrift.transport.TNonblockingTransport; import org.apache.thrift.transport.TTransportException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * This is a interim solution till THRIFT-1167 gets committed... @@ -104,7 +105,7 @@ public class CustomTHsHaServer extends TNonblockingServer public void run() { TNonblockingSocket socket = (TNonblockingSocket) frameBuffer.trans_; - SocketSessionManagementService.remoteSocket.set(socket.getSocketChannel().socket().getRemoteSocketAddress()); + ThriftSessionManager.instance.setCurrentSocket(socket.getSocketChannel().socket().getRemoteSocketAddress()); frameBuffer.invoke(); // this is how we let the same selector thread change the selection type. thread.requestSelectInterestChange(frameBuffer); diff --git a/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java b/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java index b6f76a79c2..62da846668 100644 --- a/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java +++ b/src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java @@ -19,7 +19,7 @@ package org.apache.cassandra.thrift; import java.net.InetSocketAddress; -import org.apache.cassandra.service.SocketSessionManagementService; +import org.apache.cassandra.service.ThriftSessionManager; import org.apache.thrift.server.TNonblockingServer; import org.apache.thrift.server.TServer; import org.apache.thrift.transport.TNonblockingServerTransport; @@ -37,7 +37,7 @@ public class CustomTNonBlockingServer extends TNonblockingServer protected boolean requestInvoke(FrameBuffer frameBuffer) { TNonblockingSocket socket = (TNonblockingSocket) frameBuffer.trans_; - SocketSessionManagementService.remoteSocket.set(socket.getSocketChannel().socket().getRemoteSocketAddress()); + ThriftSessionManager.instance.setCurrentSocket(socket.getSocketChannel().socket().getRemoteSocketAddress()); frameBuffer.invoke(); return true; } diff --git a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java index 00f0444195..24010b3149 100644 --- a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java +++ b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java @@ -18,6 +18,7 @@ package org.apache.cassandra.thrift; import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.net.SocketTimeoutException; import java.util.concurrent.ExecutorService; import java.util.concurrent.SynchronousQueue; @@ -28,10 +29,9 @@ import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.ThriftSessionManager; import org.apache.thrift.TException; import org.apache.thrift.TProcessor; import org.apache.thrift.protocol.TProtocol; @@ -179,8 +179,11 @@ public class CustomTThreadPoolServer extends TServer TTransport outputTransport = null; TProtocol inputProtocol = null; TProtocol outputProtocol = null; + SocketAddress socket = null; try { + socket = ((TCustomSocket) client_).getSocket().getRemoteSocketAddress(); + ThriftSessionManager.instance.setCurrentSocket(socket); processor = processorFactory_.getProcessor(client_); inputTransport = inputTransportFactory_.getTransport(client_); outputTransport = outputTransportFactory_.getTransport(client_); @@ -213,6 +216,8 @@ public class CustomTThreadPoolServer extends TServer finally { activeClients.decrementAndGet(); + if (socket != null) + ThriftSessionManager.instance.connectionComplete(socket); } if (inputTransport != null) @@ -250,34 +255,13 @@ public class CustomTThreadPoolServer extends TServer .inputProtocolFactory(args.tProtocolFactory) .outputProtocolFactory(args.tProtocolFactory) .processor(args.processor); - ExecutorService executorService = new CleaningThreadPool(args.cassandraServer.clientState, serverArgs.minWorkerThreads, serverArgs.maxWorkerThreads); + ExecutorService executorService = new ThreadPoolExecutor(serverArgs.minWorkerThreads, + serverArgs.maxWorkerThreads, + 60, + TimeUnit.SECONDS, + new SynchronousQueue(), + new NamedThreadFactory("Thrift")); return new CustomTThreadPoolServer(serverArgs, executorService); } } - - /** - * A subclass of Java's ThreadPoolExecutor which implements Jetty's ThreadPool - * interface (for integration with Avro), and performs ClientState cleanup. - * - * (Note that the tasks being executed perform their own while-command-process - * loop until the client disconnects.) - */ - private static class CleaningThreadPool extends ThreadPoolExecutor - { - private final ThreadLocal state; - - public CleaningThreadPool(ThreadLocal state, int minWorkerThread, int maxWorkerThreads) - { - super(minWorkerThread, maxWorkerThreads, 60, TimeUnit.SECONDS, new SynchronousQueue(), new NamedThreadFactory("Thrift")); - this.state = state; - } - - @Override - protected void afterExecute(Runnable r, Throwable t) - { - super.afterExecute(r, t); - DebuggableThreadPoolExecutor.logExceptionsAfterExecute(r, t); - state.get().logout(); - } - } } diff --git a/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java b/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java index 77db30ca2e..13a1dd0172 100644 --- a/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java +++ b/src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java @@ -21,7 +21,7 @@ import java.net.InetSocketAddress; import java.net.Socket; import java.net.SocketException; -import org.apache.cassandra.service.SocketSessionManagementService; +import org.apache.cassandra.service.ThriftSessionManager; import org.apache.thrift.transport.TNonblockingServerSocket; import org.apache.thrift.transport.TNonblockingSocket; import org.apache.thrift.transport.TTransportException; @@ -50,12 +50,13 @@ public class TCustomNonblockingServerSocket extends TNonblockingServerSocket if (tsocket == null || tsocket.getSocketChannel() == null) return tsocket; Socket socket = tsocket.getSocketChannel().socket(); - // clean up the old information. - SocketSessionManagementService.instance.remove(socket.getRemoteSocketAddress()); + // Any existing connection we had from this remote socket must be done now, so reset it + ThriftSessionManager.instance.connectionComplete(socket.getRemoteSocketAddress()); try { socket.setKeepAlive(this.keepAlive); - } catch (SocketException se) + } + catch (SocketException se) { logger.warn("Failed to set keep-alive on Thrift socket.", se); }