CJDBCversion1.0rc6: org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread class Reference

00160   {
00161     ArrayList vdbActiveThreads = vdb.getActiveThreads();
00162     ArrayList vdbPendingQueue = vdb.getPendingConnections();
00163     // List of open ResultSets for streaming. This is not synchronized since the
00164     // connection does only handle one request at a time
00165     HashMap streamedResultSet = new HashMap();
00166     boolean isActive = true;
00167 
00168     if (vdbActiveThreads == null)
00169     {
00170       logger
00171           .error("Got null active threads queue in VirtualDatabaseWorkerThread");
00172       isKilled = true;
00173     }
00174     if (vdbPendingQueue == null)
00175     {
00176       logger.error("Got null connection queue in VirtualDatabaseWorkerThread");
00177       isKilled = true;
00178     }
00179 
00180     // Main loop
00181     while (!isKilled)
00182     {
00183       // Get a connection from the pending queue
00184       synchronized (vdbPendingQueue)
00185       {
00186         while (vdbPendingQueue.isEmpty())
00187         {
00188           if (!vdb.poolConnectionThreads)
00189           { // User does not want thread pooling, kill this thread!
00190             isKilled = true;
00191             break;
00192           }
00193           boolean timeout = false;
00194           try
00195           {
00196             if (isActive)
00197             {
00198               isActive = false;
00199               // Remove ourselves from the active thread list
00200               synchronized (vdbActiveThreads)
00201               {
00202                 vdbActiveThreads.remove(this);
00203                 vdb.addIdleThread();
00204               }
00205             }
00206             long before = System.currentTimeMillis();
00207             vdbPendingQueue.wait(vdb.getMaxThreadIdleTime());
00208             long now = System.currentTimeMillis();
00209             // Check if timeout has expired
00210             timeout = now - before >= vdb.getMaxThreadIdleTime();
00211           }
00212           catch (InterruptedException e)
00213           {
00214             logger.warn("VirtualDatabaseWorkerThread wait() interrupted");
00215           }
00216           if (timeout && vdbPendingQueue.isEmpty())
00217           {
00218             if (vdb.currentNbOfThreads > vdb.minNbOfThreads)
00219             { // We have enough threads, kill this one
00220               isKilled = true;
00221               break;
00222             }
00223           }
00224         }
00225 
00226         if (isKilled)
00227         { // Cleaning up
00228           synchronized (vdbActiveThreads)
00229           { // Remove ourselves from the appropriate thread list
00230             if (isActive)
00231             {
00232               vdbActiveThreads.remove(this);
00233               vdb.removeCurrentNbOfThread();
00234             }
00235             else
00236               vdb.removeIdleThread();
00237           }
00238           // Get out of the while loop
00239           continue;
00240         }
00241 
00242         // Get a connection
00243         try
00244         {
00245           in = (CJDBCInputStream) vdbPendingQueue.remove(0);
00246           out = (CJDBCOutputStream) vdbPendingQueue.remove(0);
00247         }
00248         catch (Exception e)
00249         {
00250           logger.error("Error while getting streams from connection");
00251           continue;
00252         }
00253 
00254         synchronized (vdbActiveThreads)
00255         {
00256           if (!isActive)
00257           {
00258             vdb.removeIdleThread();
00259             isActive = true;
00260             // Add this thread to the active thread list
00261             vdbActiveThreads.add(this);
00262           }
00263         }
00264       }
00265 
00266       // Handle connection
00267       // Read the user information and check authentication
00268       boolean success = false;
00269       String login;
00270       try
00271       {
00272         login = in.readUTF();
00273         String password = in.readUTF();
00274         user = new VirtualDatabaseUser(login, password);
00275 
00276         if (vdb.getAuthenticationManager().isValidVirtualUser(user))
00277         { // Authentication ok, send if SQL skeleton is needed
00278           boolean needSkeleton = vdb.getRequestManager()
00279               .getRequiredParsingGranularity() != ParsingGranularities.NO_PARSING;
00280           out.writeObject(new Boolean(needSkeleton));
00281           out.writeObject(vdb.getBlobFilter());
00282           out.flush();
00283           success = true;
00284 
00285           if (logger.isDebugEnabled())
00286             logger.debug("Login accepted for " + login);
00287         }
00288         else
00289         { // Authentication failed, close the connection
00290           if (logger.isDebugEnabled())
00291             logger.debug("Authentication failed for user '" + login + "'");
00292           continue;
00293         }
00294       }
00295       catch (OptionalDataException e)
00296       {
00297         logger.error("Protocol error while expecting user authentication (" + e
00298             + ")");
00299         continue;
00300       }
00301       catch (IOException e)
00302       {
00303         logger.error("I/O error during user authentication (" + e + ")");
00304         continue;
00305       }
00306       finally
00307       {
00308         if (!success)
00309         {
00310           try
00311           {
00312             out.close();
00313             in.close();
00314           }
00315           catch (IOException ignore)
00316           {
00317           }
00318         }
00319       }
00320 
00321       // Ok, let's process the queries on this connection
00322       long currentTid = 0;
00323       boolean transactionStarted = false;
00324       boolean closed = false;
00325       int objectsOnStream = 0;
00326       int command;
00327       while (!closed && !isKilled)
00328       {
00329         try
00330         {
00331           // Get the query
00332           command = in.readInt();
00333 
00334           // Process it
00335           switch (command)
00336           {
00337             case Commands.ExecReadRequest :
00338               if (logger.isDebugEnabled())
00339                 logger.debug("ExecReadRequest command");
00340               SelectRequest select = (SelectRequest) in.readObject();
00341               transactionStarted = setRequestParameters(select, login,
00342                   currentTid, transactionStarted);
00343               if (!transactionStarted)
00344                 currentTid = 0;
00345               ControllerResultSet crs = vdb.execReadRequest(select);
00346               if (crs.hasMoreData())
00347                 streamedResultSet.put(crs.getCursorName(), crs);
00348               out.writeObject(crs.getFields());
00349               out.writeObject(crs.getData());
00350               out.writeBoolean(crs.hasMoreData());
00351               if (crs.hasMoreData())
00352                 out.writeUTF(crs.getCursorName());
00353               out.flush();
00354               objectsOnStream++;
00355               break;
00356             case Commands.ExecWriteRequest :
00357               if (logger.isDebugEnabled())
00358                 logger.debug("ExecWriteRequest command");
00359               AbstractWriteRequest write = (AbstractWriteRequest) in
00360                   .readObject();
00361               transactionStarted = setRequestParameters(write, login,
00362                   currentTid, transactionStarted);
00363               if (!transactionStarted)
00364                 currentTid = 0;
00365               out.writeObject(new Integer(vdb.execWriteRequest(write)));
00366               out.flush();
00367               objectsOnStream++;
00368               break;
00369             case Commands.ExecWriteRequestWithKeys :
00370               if (logger.isDebugEnabled())
00371                 logger.debug("ExecWriteRequestWithKeys command");
00372               AbstractWriteRequest writeWithKeys = (AbstractWriteRequest) in
00373                   .readObject();
00374               transactionStarted = setRequestParameters(writeWithKeys, login,
00375                   currentTid, transactionStarted);
00376               if (!transactionStarted)
00377                 currentTid = 0;
00378               ControllerResultSet keys = vdb
00379                   .execWriteRequestWithKeys(writeWithKeys);
00380               out.writeObject(keys.getFields());
00381               out.writeObject(keys.getData());
00382               out.writeBoolean(keys.hasMoreData());
00383               if (keys.hasMoreData())
00384                 out.writeUTF(keys.getCursorName());
00385               out.flush();
00386               objectsOnStream++;
00387               break;
00388             case Commands.ExecReadStoredProcedure :
00389               if (logger.isDebugEnabled())
00390                 logger.debug("ExecReadStoredProcedure command");
00391               StoredProcedure readProc = (StoredProcedure) in.readObject();
00392               transactionStarted = setRequestParameters(readProc, login,
00393                   currentTid, transactionStarted);
00394               if (!transactionStarted)
00395                 currentTid = 0;
00396               ControllerResultSet sprs = vdb.execReadStoredProcedure(readProc);
00397               out.writeObject(sprs.getFields());
00398               out.writeObject(sprs.getData());
00399               out.writeBoolean(sprs.hasMoreData());
00400               if (sprs.hasMoreData())
00401                 out.writeUTF(sprs.getCursorName());
00402               out.flush();
00403               objectsOnStream++;
00404               break;
00405             case Commands.ExecWriteStoredProcedure :
00406               if (logger.isDebugEnabled())
00407                 logger.debug("ExecWriteStoredProcedure command");
00408               StoredProcedure writeProc = (StoredProcedure) in.readObject();
00409               transactionStarted = setRequestParameters(writeProc, login,
00410                   currentTid, transactionStarted);
00411               if (!transactionStarted)
00412                 currentTid = 0;
00413               out.writeObject(new Integer(vdb
00414                   .execWriteStoredProcedure(writeProc)));
00415               out.flush();
00416               objectsOnStream++;
00417               break;
00418             case Commands.Begin :
00419               if (logger.isDebugEnabled())
00420                 logger.debug("Begin command");
00421               currentTid = vdb.begin(login);
00422               out.writeObject(new Long(currentTid));
00423               out.flush();
00424               transactionStarted = true;
00425               break;
00426             case Commands.Commit :
00427               if (logger.isDebugEnabled())
00428                 logger.debug("Commit command");
00429               vdb.commit(currentTid);
00430               currentTid = vdb.begin(login);
00431               out.writeObject(new Long(currentTid));
00432               out.flush();
00433               if (objectsOnStream > 0)
00434               {
00435                 // Reset the outputstream else sent ResultSets cannot be GC
00436                 out.reset();
00437                 objectsOnStream = 0;
00438               }
00439               break;
00440             case Commands.Rollback :
00441               if (logger.isDebugEnabled())
00442                 logger.debug("Rollback command");
00443               vdb.rollback(currentTid);
00444               currentTid = vdb.begin(login);
00445               out.writeObject(new Long(currentTid));
00446               out.flush();
00447               if (objectsOnStream > 0)
00448               {
00449                 // Reset the outputstream else sent ResultSets cannot be GC
00450                 out.reset();
00451                 objectsOnStream = 0;
00452               }
00453               break;
00454             case Commands.GetVirtualDatabaseName :
00455               if (logger.isDebugEnabled())
00456                 logger.debug("GetVirtualDatabaseName command");
00457               out.writeObject(vdb.getDatabaseName());
00458               out.flush();
00459               break;
00460             case Commands.GetDatabaseProductName :
00461               if (logger.isDebugEnabled())
00462                 logger.debug("GetDatabaseProductName command");
00463               out.writeObject(vdb.getDatabaseProductName());
00464               out.flush();
00465               break;
00466             case Commands.GetControllerVersionNumber :
00467               if (logger.isDebugEnabled())
00468                 logger.debug("GetControllerVersionNumber command");
00469               out.writeObject(Constants.VERSION);
00470               out.flush();
00471               break;
00472             case Commands.DatabaseMetaDataGetTables :
00473               if (logger.isDebugEnabled())
00474                 logger.debug("DatabaseMetaDataGetTables command");
00475               String tcatalog = (String) in.readObject();
00476               String tschemaPattern = (String) in.readObject();
00477               String ttableNamePattern = (String) in.readObject();
00478               String[] ttypes = (String[]) in.readObject();
00479               out.writeObject(vdb.getMetaData().getTables(tcatalog,
00480                   tschemaPattern, ttableNamePattern, ttypes));
00481               out.flush();
00482               objectsOnStream++;
00483               break;
00484             case Commands.DatabaseMetaDataGetColumns :
00485               if (logger.isDebugEnabled())
00486                 logger.debug("DatabaseMetaDataGetColumns command");
00487               String ccatalog = (String) in.readObject();
00488               String cschemaPattern = (String) in.readObject();
00489               String ctableNamePattern = (String) in.readObject();
00490               String ccolumnNamePattern = (String) in.readObject();
00491               out.writeObject(vdb.getMetaData().getColumns(ccatalog,
00492                   cschemaPattern, ctableNamePattern, ccolumnNamePattern));
00493               out.flush();
00494               objectsOnStream++;
00495               break;
00496             case Commands.DatabaseMetaDataGetPrimaryKeys :
00497               if (logger.isDebugEnabled())
00498                 logger.debug("DatabaseMetaDataGetPrimaryKeys command");
00499               String pcatalog = (String) in.readObject();
00500               String pschemaPattern = (String) in.readObject();
00501               String ptableNamePattern = (String) in.readObject();
00502               out.writeObject(vdb.getMetaData().getPrimaryKeys(pcatalog,
00503                   pschemaPattern, ptableNamePattern));
00504               out.flush();
00505               objectsOnStream++;
00506               break;
00507             case Commands.DatabaseMetaDataGetProcedures :
00508               if (logger.isDebugEnabled())
00509                 logger.debug("DatabaseMetaDataGetProcedures command");
00510               String rcatalog = (String) in.readObject();
00511               String rschemaPattern = (String) in.readObject();
00512               String procedureNamePattern = (String) in.readObject();
00513               out.writeObject(vdb.getMetaData().getProcedures(rcatalog,
00514                   rschemaPattern, procedureNamePattern));
00515               out.flush();
00516               objectsOnStream++;
00517               break;
00518             case Commands.DatabaseMetaDataGetProcedureColumns :
00519               if (logger.isDebugEnabled())
00520                 logger.debug("DatabaseMetaDataGetProcedureColumns command");
00521               String pccatalog = (String) in.readObject();
00522               String pcschemaPattern = (String) in.readObject();
00523               String pcprocedureNamePattern = (String) in.readObject();
00524               String pccolumnNamePattern = (String) in.readObject();
00525               out
00526                   .writeObject(vdb.getMetaData().getProcedureColumns(pccatalog,
00527                       pcschemaPattern, pcprocedureNamePattern,
00528                       pccolumnNamePattern));
00529               out.flush();
00530               objectsOnStream++;
00531               break;
00532             case Commands.DatabaseMetaDataGetCatalogs :
00533               if (logger.isDebugEnabled())
00534                 logger.debug("DatabaseMetaDataGetCatalogs command");
00535               ArrayList list = controller.listVirtualDatabases();
00536               out.writeObject(vdb.getMetaData().getCatalogs(list));
00537               out.flush();
00538               objectsOnStream++;
00539               break;
00540             case Commands.DatabaseMetaDataGetCatalog :
00541               if (logger.isDebugEnabled())
00542                 logger.debug("DatabaseMetaDataGetCatalog command");
00543               out.writeObject(vdb.getName());
00544               out.flush();
00545               objectsOnStream++;
00546               break;
00547             case Commands.DatabaseMetaDataGetTableTypes :
00548               if (logger.isDebugEnabled())
00549                 logger.debug("DatabaseMetaDataGetTable Types command");
00550               out.writeObject(vdb.getMetaData().getTableTypes());
00551               out.flush();
00552               objectsOnStream++;
00553               break;
00554             case Commands.DatabaseMetaDataGetSchemas :
00555               if (logger.isDebugEnabled())
00556                 logger.debug("DatabaseMetaDataGetSchemas Types command");
00557               out.writeObject(vdb.getMetaData().getSchemas());
00558               out.flush();
00559               objectsOnStream++;
00560               break;
00561             case Commands.DatabaseMetaDataGetTablePrivileges :
00562               if (logger.isDebugEnabled())
00563                 logger.debug("DatabaseMetaDataGetTablePrivileges command");
00564               String tpcatalog = (String) in.readObject();
00565               String tpschemaPattern = (String) in.readObject();
00566               String tptablePattern = (String) in.readObject();
00567               out.writeObject(vdb.getMetaData().getTablePrivileges(tpcatalog,
00568                   tpschemaPattern, tptablePattern));
00569               out.flush();
00570               objectsOnStream++;
00571               break;
00572             case Commands.ConnectionSetCatalog :
00573               if (logger.isDebugEnabled())
00574                 logger.debug("ConnectionSetCatalog command");
00575               String catalog = (String) in.readObject();
00576               boolean change = controller.hasVirtualDatabase(catalog);
00577               if (change)
00578               {
00579                 VirtualDatabase tempvdb = controller
00580                     .getVirtualDatabase(catalog);
00581                 if (!tempvdb.getAuthenticationManager()
00582                     .isValidVirtualUser(user))
00583                   out
00584                       .writeObject(new SQLException(
00585                           "User authentication has failed for asked catalog. No change"));
00586                 else
00587                 {
00588                   this.vdb = tempvdb;
00589                   out.writeObject(new Boolean(true));
00590                 }
00591               }
00592               else
00593                 out.writeObject(new Boolean(false));
00594               out.flush();
00595               objectsOnStream++;
00596               break;
00597             case Commands.Close :
00598               if (logger.isDebugEnabled())
00599                 logger.debug("Close command");
00600               out.writeObject(completed);
00601               out.flush();
00602               closed = true;
00603               break;
00604             case Commands.Reset :
00605               if (logger.isDebugEnabled())
00606                 logger.debug("Reset command");
00607 
00608               // Reset the outputstream else sent ResultSets cannot be GC
00609               out.reset();
00610 
00611               // Do the cleanup
00612               if (transactionStarted)
00613               {
00614                 if (logger.isDebugEnabled())
00615                   logger.debug("Forcing transaction " + currentTid
00616                       + " rollback");
00617                 try
00618                 {
00619                   vdb.rollback(currentTid);
00620                 }
00621                 catch (Exception e)
00622                 {
00623                   if (logger.isDebugEnabled())
00624                     logger.debug("Error during rollback of transaction "
00625                         + currentTid + "(" + e + ")");
00626                 }
00627               }
00628               currentTid = 0;
00629               transactionStarted = false;
00630               break;
00631             case Commands.FetchNextResultSetRows :
00632               if (logger.isDebugEnabled())
00633                 logger.debug("FetchNextResultSetRows command");
00634 
00635               String cursorName = in.readUTF();
00636               int fetchSize = in.readInt();
00637               ControllerResultSet fetchCrs = (ControllerResultSet) streamedResultSet
00638                   .get(cursorName);
00639               if (fetchCrs == null)
00640               {
00641                 out.writeObject(new SQLException(
00642                     "No valid ControllerResultSet to fetch data from"));
00643                 out.flush();
00644               }
00645               else
00646               {
00647                 out.writeObject(fetchCrs.fetchData());
00648                 out.writeBoolean(fetchCrs.hasMoreData());
00649                 out.flush();
00650                 if (!fetchCrs.hasMoreData())
00651                   streamedResultSet.remove(cursorName);
00652               }
00653               break;
00654             case Commands.CloseRemoteResultSet :
00655               if (logger.isDebugEnabled())
00656                 logger.debug("CloseRemoteResultSet command");
00657 
00658               String cursor = in.readUTF();
00659               ControllerResultSet crsToClose = (ControllerResultSet) streamedResultSet
00660                   .remove(cursor);
00661               if (crsToClose == null)
00662               {
00663                 out.writeObject(new SQLException(
00664                     "No valid ControllerResultSet to close."));
00665                 out.flush();
00666               }
00667               else
00668               {
00669                 crsToClose.closeResultSet();
00670                 out.writeObject(completed);
00671                 out.flush();
00672               }
00673               break;
00674             default :
00675               logger.error("Unsupported protocol command: " + command);
00676               break;
00677           }
00678 
00679           // Check if the stream needs to be reseted
00680           if (objectsOnStream > STREAM_GC_LIMIT)
00681           { // Reset the outputstream else sent ResultSets cannot be GC
00682             out.reset();
00683             objectsOnStream = 0;
00684           }
00685         }
00686         catch (OptionalDataException e)
00687         {
00688           logger.warn("Protocol error (" + e + ")");
00689           try
00690           {
00691             out.writeObject(e);
00692             out.flush();
00693           }
00694           catch (IOException ignore)
00695           {
00696           }
00697         }
00698         catch (RemoteException e)
00699         {
00700           logger.warn("Error during command execution (" + e + ")");
00701           try
00702           {
00703             out.writeObject(e);
00704             out.flush();
00705           }
00706           catch (IOException ignore)
00707           {
00708           }
00709         }
00710         catch (EOFException e)
00711         {
00712           logger.warn("Client " + login + " closed connection with server");
00713           closed = true;
00714         }
00715         catch (IOException e)
00716         {
00717           logger.warn("Closing connection with client " + login
00718               + " because of IOException.(" + e + ")");
00719           closed = true;
00720         }
00721         catch (ClassNotFoundException e)
00722         {
00723           logger.error("Protocol error (" + e + ")");
00724           try
00725           {
00726             out.writeObject(e);
00727             out.flush();
00728           }
00729           catch (IOException ignore)
00730           {
00731           }
00732         }
00733         catch (SQLException e)
00734         {
00735           logger
00736               .warn("Error during command execution (" + e.getMessage() + ")");
00737           try
00738           {
00739             out.writeObject(e);
00740             out.flush();
00741           }
00742           catch (IOException ignore)
00743           {
00744           }
00745         }
00746         catch (RuntimeException e)
00747         {
00748           logger.warn("Runtime error during command execution ("
00749               + e.getMessage() + ")");
00750           try
00751           {
00752             out.writeObject(new SQLException(e.getMessage()));
00753             out.flush();
00754           }
00755           catch (IOException ignore)
00756           {
00757           }
00758         }
00759       }
00760 
00761       // Do the cleanup
00762       if (transactionStarted)
00763       {
00764         if (logger.isDebugEnabled())
00765           logger.debug("Forcing transaction " + currentTid + " rollback");
00766         try
00767         {
00768           vdb.rollback(currentTid);
00769         }
00770         catch (Exception e)
00771         {
00772           if (logger.isDebugEnabled())
00773             logger.debug("Error during rollback of transaction " + currentTid
00774                 + "(" + e + ")");
00775         }
00776       }
00777       if (!streamedResultSet.isEmpty())
00778       {
00779         for (Iterator iter = streamedResultSet.values().iterator(); iter
00780             .hasNext();)
00781         {
00782           ControllerResultSet crs = (ControllerResultSet) iter.next();
00783           crs.closeResultSet();
00784 
00785         }
00786         streamedResultSet.clear();
00787       }
00788       try
00789       {
00790         in.close();
00791       }
00792       catch (IOException ignore)
00793       {
00794       }
00795       try
00796       {
00797         out.close();
00798       }
00799       catch (IOException ignore)
00800       {
00801       }
00802     }
00803 
00804     if (logger.isDebugEnabled())
00805       logger.debug("VirtualDatabaseWorkerThread terminating.");
00806   }

Public メソッド
	VirtualDatabaseWorkerThread (Controller controller, VirtualDatabase vdb)
void	shutdown ()
void	run ()
String[]	retrieveClientData ()
long	getBytesRead ()
long	getBytesWritten ()
long	getTimeActive ()
long	getReadingSpeed ()
long	getWritingSpeed ()
String	getUser ()
Private メソッド
boolean	setRequestParameters (AbstractRequest request, String login, long tid, boolean transactionStarted) throws SQLException
Private 変数
boolean	isKilled = false
VirtualDatabase	vdb
final CommandCompleted	completed = new CommandCompleted()
Trace	logger = null
CJDBCInputStream	in = null
CJDBCOutputStream	out = null
VirtualDatabaseUser	user
Controller	controller
Static Private 変数
final int	STREAM_GC_LIMIT = 5
int	threadID = 0
クラス org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread

説明

Public メソッド

Private メソッド

Private 変数

Static Private 変数

コンストラクタとデストラクタ

メソッド

変数
