natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

clean up Thrift session management and encapsulate in ThriftSessionManager 

patch by jbellis; reviewed by dbrosius for CASSANDRA-4657
This commit is contained in:
Jonathan Ellis 2012-10-08 11:04:11 -05:00
parent fa56af1336
commit e8438b8084
7 changed files with 36 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/java/org/apache/cassandra/service/ClientState.java
View File

@ -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<Integer, CQLStatement> 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);

59
src/java/org/apache/cassandra/service/SocketSessionManagementService.java
View File

@ -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<SocketAddress> remoteSocket = new ThreadLocal<SocketAddress>();
private final Map<SocketAddress, ClientState> activeSocketSessions = new ConcurrentHashMap<SocketAddress, ClientState>();
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();
}
}

31
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -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<Column> EMPTY_SUBCOLUMNS = Collections.emptyList();
private final static List<CounterColumn> EMPTY_COUNTER_SUBCOLUMNS = Collections.emptyList();
// thread local state containing session information
public final ThreadLocal<ClientState> clientState = new ThreadLocal<ClientState>()
{
@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<DecoratedKey, ColumnFamily> readColumnFamily(List<ReadCommand> commands, org.apache.cassandra.db.ConsistencyLevel consistency_level)

9
src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java
View File

@ -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);

4
src/java/org/apache/cassandra/thrift/CustomTNonBlockingServer.java
View File

@ -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;
}

42
src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java
View File

@ -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<Runnable>(),
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<ClientState> state;
public CleaningThreadPool(ThreadLocal<ClientState> state, int minWorkerThread, int maxWorkerThreads)
{
super(minWorkerThread, maxWorkerThreads, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), 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();
}
}
}

9
src/java/org/apache/cassandra/thrift/TCustomNonblockingServerSocket.java
View File

@ -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);
}