ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject ignite git commit: IGNITE-2643: Fixed.
Date Thu, 18 Feb 2016 13:54:56 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-1786 fe63a9086 -> 6a7cb0a9c


IGNITE-2643: Fixed.


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/6a7cb0a9
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/6a7cb0a9
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/6a7cb0a9

Branch: refs/heads/ignite-1786
Commit: 6a7cb0a9c242708c0e267465e812a0ca90bd8459
Parents: fe63a90
Author: vozerov-gridgain <vozerov@gridgain.com>
Authored: Thu Feb 18 16:54:45 2016 +0300
Committer: vozerov-gridgain <vozerov@gridgain.com>
Committed: Thu Feb 18 16:54:45 2016 +0300

----------------------------------------------------------------------
 .../ignite/configuration/OdbcConfiguration.java | 27 +++++++
 .../processors/odbc/OdbcNioListener.java        | 76 +++++---------------
 .../internal/processors/odbc/OdbcProcessor.java |  2 +-
 .../processors/odbc/OdbcRequestHandler.java     | 69 ++++++++++++------
 4 files changed, 95 insertions(+), 79 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/6a7cb0a9/modules/core/src/main/java/org/apache/ignite/configuration/OdbcConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/OdbcConfiguration.java
b/modules/core/src/main/java/org/apache/ignite/configuration/OdbcConfiguration.java
index d6f0500..8f0a0fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/OdbcConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/OdbcConfiguration.java
@@ -38,6 +38,9 @@ public class OdbcConfiguration {
     /** Default socket send and receive buffer size. */
     public static final int DFLT_SOCK_BUF_SIZE = 32 * 1024;
 
+    /** Default max number of open cursors per connection. */
+    public static final int DFLT_MAX_OPEN_CURSORS = 128;
+
     /** TCP port. */
     private int port = DFLT_TCP_PORT;
 
@@ -65,6 +68,9 @@ public class OdbcConfiguration {
     /** Idle timeout. */
     private long idleTimeout = DFLT_IDLE_TIMEOUT;
 
+    /** Max number of opened cursors per connection. */
+    private int maxOpenCursors = DFLT_MAX_OPEN_CURSORS;
+
     /**
      * Creates ODBC server configuration with all default values.
      */
@@ -84,6 +90,7 @@ public class OdbcConfiguration {
         directBuf = cfg.isDirectBuffer();
         host = cfg.getHost();
         idleTimeout = cfg.getIdleTimeout();
+        maxOpenCursors = cfg.getMaxOpenCursors();
         noDelay = cfg.isNoDelay();
         port = cfg.getPort();
         rcvBufSize = cfg.getReceiveBufferSize();
@@ -289,4 +296,24 @@ public class OdbcConfiguration {
     public void setIdleTimeout(long idleTimeout) {
         this.idleTimeout = idleTimeout;
     }
+
+    /**
+     * Gets maximum number of opened cursors per connection.
+     * <p>
+     * Defaults to {@link #DFLT_MAX_OPEN_CURSORS}.
+     *
+     * @return Maximum number of opened cursors.
+     */
+    public int getMaxOpenCursors() {
+        return maxOpenCursors;
+    }
+
+    /**
+     * Sets maximum number of opened cursors per connection. See {@link #getMaxOpenCursors()}.
+     *
+     * @param maxOpenCursors Maximum number of opened cursors.
+     */
+    public void setMaxOpenCursors(int maxOpenCursors) {
+        this.maxOpenCursors = maxOpenCursors;
+    }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/6a7cb0a9/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcNioListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcNioListener.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcNioListener.java
index 23560b1..e25ae5b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcNioListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcNioListener.java
@@ -29,6 +29,7 @@ import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProce
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.nio.GridNioServerListenerAdapter;
 import org.apache.ignite.internal.util.nio.GridNioSession;
+import org.apache.ignite.internal.util.nio.GridNioSessionMetaKey;
 import org.jetbrains.annotations.Nullable;
 
 import java.io.IOException;
@@ -42,14 +43,17 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
     /** Initial output stream capacity. */
     private static final int INIT_CAP = 1024;
 
+    /** Handler metadata key. */
+    private static final int HANDLER_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
+
     /** Request ID generator. */
     private static final AtomicLong REQ_ID_GEN = new AtomicLong();
 
     /** Busy lock. */
     private final GridSpinBusyLock busyLock;
 
-    /** Request handler. */
-    private final OdbcRequestHandler handler;
+    /** Kernal context. */
+    private final GridKernalContext ctx;
 
     /** Marshaller. */
     private final GridBinaryMarshaller marsh;
@@ -60,15 +64,14 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
     /**
      * @param ctx Context.
      * @param busyLock Shutdown busy lock.
-     * @param handler Request handler.
      */
-    public OdbcNioListener(GridKernalContext ctx, GridSpinBusyLock busyLock, OdbcRequestHandler
handler) {
+    public OdbcNioListener(GridKernalContext ctx, GridSpinBusyLock busyLock) {
+        this.ctx = ctx;
         this.busyLock = busyLock;
-        this.handler = handler;
 
         CacheObjectBinaryProcessorImpl cacheObjProc = (CacheObjectBinaryProcessorImpl)ctx.cacheObjects();
 
-        marsh = cacheObjProc.marshaller();
+        this.marsh = cacheObjProc.marshaller();
 
         this.log = ctx.log(getClass());
     }
@@ -77,6 +80,8 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
     @Override public void onConnected(GridNioSession ses) {
         if (log.isDebugEnabled())
             log.debug("ODBC client connected: " + ses.remoteAddress());
+
+        ses.addMeta(HANDLER_META_KEY, new OdbcRequestHandler(ctx, busyLock));
     }
 
     /** {@inheritDoc} */
@@ -102,10 +107,15 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
             if (log.isDebugEnabled()) {
                 startTime = System.nanoTime();
 
-                log.debug("ODBC request received [id=" + reqId + ", addr=" + ses.remoteAddress()
+ ", req=" + req + ']');
+                log.debug("ODBC request received [id=" + reqId + ", addr=" + ses.remoteAddress()
+
+                    ", req=" + req + ']');
             }
 
-            OdbcResponse resp = handle(req);
+            OdbcRequestHandler handler = ses.meta(HANDLER_META_KEY);
+
+            assert handler != null;
+
+            OdbcResponse resp = handler.handle(req);
 
             if (log.isDebugEnabled()) {
                 long dur = (System.nanoTime() - startTime) / 1000;
@@ -160,16 +170,10 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
 
         switch (cmd) {
             case OdbcRequest.EXECUTE_SQL_QUERY: {
-
                 String cache = reader.readString();
                 String sql = reader.readString();
                 int argsNum = reader.readInt();
 
-                if (log.isDebugEnabled()) {
-                    log.debug("Message: [cmd=EXECUTE_SQL_QUERY, cache=" + cache +
-                            ", query=" + sql + ", argsNum=" + argsNum + ']');
-                }
-
                 Object[] params = new Object[argsNum];
 
                 for (int i = 0; i < argsNum; ++i)
@@ -181,26 +185,17 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
             }
 
             case OdbcRequest.FETCH_SQL_QUERY: {
-
                 long queryId = reader.readLong();
                 int pageSize = reader.readInt();
 
-                if (log.isDebugEnabled())
-                    log.debug("Message: [cmd=FETCH_SQL_QUERY, queryId=" + queryId + ", pageSize="
+ pageSize + ']');
-
                 res = new OdbcQueryFetchRequest(queryId, pageSize);
 
                 break;
             }
 
             case OdbcRequest.CLOSE_SQL_QUERY: {
-
                 long queryId = reader.readLong();
 
-                if (log.isDebugEnabled()) {
-                    log.debug("Message: [cmd=CLOSE_SQL_QUERY, queryId=" + queryId + ']');
-                }
-
                 res = new OdbcQueryCloseRequest(queryId);
 
                 break;
@@ -212,36 +207,24 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
                 String table = reader.readString();
                 String column = reader.readString();
 
-                if (log.isDebugEnabled()) {
-                    log.debug("Message: [cmd=GET_COLUMNS_META, cache=" + cache +
-                            ", table=" + table + ", column: " + column + ']');
-                }
-
                 res = new OdbcQueryGetColumnsMetaRequest(cache, table, column);
 
                 break;
             }
 
             case OdbcRequest.GET_TABLES_META: {
-
                 String catalog = reader.readString();
                 String schema = reader.readString();
                 String table = reader.readString();
                 String tableType = reader.readString();
 
-                if (log.isDebugEnabled()) {
-                    log.debug("Message: [cmd=GET_COLUMNS_META, catalog=" + catalog +
-                            ", schema=" + schema + ", table=" + table + ", tableType=" +
tableType + ']');
-                }
-
                 res = new OdbcQueryGetTablesMetaRequest(catalog, schema, table, tableType);
 
                 break;
             }
 
             default:
-                throw new IOException("Failed to parse incoming packet (unknown command type)
" +
-                        "[cmd=[" + Byte.toString(cmd) + ']');
+                throw new IOException("Unknown ODBC command: " + cmd);
         }
 
         return res;
@@ -352,25 +335,4 @@ public class OdbcNioListener extends GridNioServerListenerAdapter<byte[]>
{
 
         return writer.array();
     }
-
-    /**
-     * Handle request.
-     *
-     * @param req Request.
-     * @return Response.
-     */
-    private OdbcResponse handle(OdbcRequest req) {
-        assert req != null;
-
-        if (!busyLock.enterBusy())
-            return new OdbcResponse(OdbcResponse.STATUS_FAILED,
-                "Failed to handle ODBC request because node is stopping: " + req);
-
-        try {
-            return handler.handle(req);
-        }
-        finally {
-            busyLock.leaveBusy();
-        }
-    }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/6a7cb0a9/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcProcessor.java
index 9d22e4a..87be686 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcProcessor.java
@@ -73,7 +73,7 @@ public class OdbcProcessor extends GridProcessorAdapter {
                 srv = GridNioServer.<byte[]>builder()
                     .address(host)
                     .port(port)
-                    .listener(new OdbcNioListener(ctx, busyLock, new OdbcRequestHandler(ctx)))
+                    .listener(new OdbcNioListener(ctx, busyLock))
                     .logger(log)
                     .selectorCount(odbcCfg.getSelectorCount())
                     .gridName(ctx.gridName())

http://git-wip-us.apache.org/repos/asf/ignite/blob/6a7cb0a9/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcRequestHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcRequestHandler.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcRequestHandler.java
index 332a5cc..1af14b3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcRequestHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/OdbcRequestHandler.java
@@ -20,10 +20,12 @@ package org.apache.ignite.internal.processors.odbc;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.cache.query.QueryCursor;
 import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.configuration.OdbcConfiguration;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.processors.cache.QueryCursorImpl;
 import org.apache.ignite.internal.processors.query.GridQueryFieldMetadata;
 import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
+import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.lang.IgniteBiTuple;
 
@@ -43,16 +45,21 @@ public class OdbcRequestHandler {
     /** Kernel context. */
     private final GridKernalContext ctx;
 
+    /** Busy lock. */
+    private final GridSpinBusyLock busyLock;
+
     /** Current queries cursors. */
-    private final ConcurrentHashMap<Long, IgniteBiTuple<QueryCursor, Iterator>>
qryCurs = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<Long, IgniteBiTuple<QueryCursor, Iterator>>
qryCursors = new ConcurrentHashMap<>();
 
     /**
      * Constructor.
      *
      * @param ctx Context.
+     * @param busyLock Shutdown latch.
      */
-    public OdbcRequestHandler(final GridKernalContext ctx) {
+    public OdbcRequestHandler(final GridKernalContext ctx, final GridSpinBusyLock busyLock)
{
         this.ctx = ctx;
+        this.busyLock = busyLock;
     }
 
     /**
@@ -64,24 +71,33 @@ public class OdbcRequestHandler {
     public OdbcResponse handle(OdbcRequest req) {
         assert req != null;
 
-        switch (req.command()) {
-            case EXECUTE_SQL_QUERY:
-                return executeQuery((OdbcQueryExecuteRequest)req);
+        if (!busyLock.enterBusy())
+            return new OdbcResponse(OdbcResponse.STATUS_FAILED,
+                    "Failed to handle ODBC request because node is stopping: " + req);
 
-            case FETCH_SQL_QUERY:
-                return fetchQuery((OdbcQueryFetchRequest)req);
+        try {
+            switch (req.command()) {
+                case EXECUTE_SQL_QUERY:
+                    return executeQuery((OdbcQueryExecuteRequest)req);
 
-            case CLOSE_SQL_QUERY:
-                return closeQuery((OdbcQueryCloseRequest)req);
+                case FETCH_SQL_QUERY:
+                    return fetchQuery((OdbcQueryFetchRequest)req);
 
-            case GET_COLUMNS_META:
-                return getColumnsMeta((OdbcQueryGetColumnsMetaRequest) req);
+                case CLOSE_SQL_QUERY:
+                    return closeQuery((OdbcQueryCloseRequest)req);
 
-            case GET_TABLES_META:
-                return getTablesMeta((OdbcQueryGetTablesMetaRequest) req);
-        }
+                case GET_COLUMNS_META:
+                    return getColumnsMeta((OdbcQueryGetColumnsMetaRequest) req);
 
-        return new OdbcResponse(OdbcResponse.STATUS_FAILED, "Unsupported ODBC request: "
+ req);
+                case GET_TABLES_META:
+                    return getTablesMeta((OdbcQueryGetTablesMetaRequest) req);
+            }
+
+            return new OdbcResponse(OdbcResponse.STATUS_FAILED, "Unsupported ODBC request:
" + req);
+        }
+        finally {
+            busyLock.leaveBusy();
+        }
     }
 
     /**
@@ -91,6 +107,17 @@ public class OdbcRequestHandler {
      * @return Response.
      */
     private OdbcResponse executeQuery(OdbcQueryExecuteRequest req) {
+        OdbcConfiguration cfg = ctx.config().getOdbcConfiguration();
+
+        assert cfg != null;
+
+        int cursorCnt = qryCursors.size();
+
+        if (cursorCnt >= cfg.getMaxOpenCursors())
+            return new OdbcResponse(OdbcResponse.STATUS_FAILED, "Too many opened cursors
(either close other " +
+                "opened cursors or increase the limit through OdbcConfiguration.setMaxOpenCursors())
" +
+                "[maximum=" + cfg.getMaxOpenCursors() + ", current=" + cursorCnt + ']');
+
         long qryId = QRY_ID_GEN.getAndIncrement();
 
         try {
@@ -108,7 +135,7 @@ public class OdbcRequestHandler {
 
             Iterator iter = qryCur.iterator();
 
-            qryCurs.put(qryId, new IgniteBiTuple<>(qryCur, iter));
+            qryCursors.put(qryId, new IgniteBiTuple<>(qryCur, iter));
 
             List<?> fieldsMeta = ((QueryCursorImpl) qryCur).fieldsMeta();
 
@@ -117,7 +144,7 @@ public class OdbcRequestHandler {
             return new OdbcResponse(res);
         }
         catch (Exception e) {
-            qryCurs.remove(qryId);
+            qryCursors.remove(qryId);
 
             return new OdbcResponse(OdbcResponse.STATUS_FAILED, e.getMessage());
         }
@@ -131,21 +158,21 @@ public class OdbcRequestHandler {
      */
     private OdbcResponse closeQuery(OdbcQueryCloseRequest req) {
         try {
-            QueryCursor cur = qryCurs.get(req.queryId()).get1();
+            QueryCursor cur = qryCursors.get(req.queryId()).get1();
 
             if (cur == null)
                 return new OdbcResponse(OdbcResponse.STATUS_FAILED, "Failed to find query
with ID: " + req.queryId());
 
             cur.close();
 
-            qryCurs.remove(req.queryId());
+            qryCursors.remove(req.queryId());
 
             OdbcQueryCloseResult res = new OdbcQueryCloseResult(req.queryId());
 
             return new OdbcResponse(res);
         }
         catch (Exception e) {
-            qryCurs.remove(req.queryId());
+            qryCursors.remove(req.queryId());
 
             return new OdbcResponse(OdbcResponse.STATUS_FAILED, e.getMessage());
         }
@@ -159,7 +186,7 @@ public class OdbcRequestHandler {
      */
     private OdbcResponse fetchQuery(OdbcQueryFetchRequest req) {
         try {
-            Iterator cur = qryCurs.get(req.queryId()).get2();
+            Iterator cur = qryCursors.get(req.queryId()).get2();
 
             if (cur == null)
                 return new OdbcResponse(OdbcResponse.STATUS_FAILED, "Failed to find query
with ID: " + req.queryId());


Mime
View raw message