ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject incubator-ignite git commit: # gg-10561: reimplement session indexing
Date Fri, 24 Jul 2015 13:45:39 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-gg-10561 dbe2798cb -> cb51025fe


# gg-10561: reimplement session indexing


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

Branch: refs/heads/ignite-gg-10561
Commit: cb51025fe590be5df85205c6160d399b2411f0a0
Parents: dbe2798
Author: ashutak <ashutak@gridgain.com>
Authored: Fri Jul 24 16:46:09 2015 +0300
Committer: ashutak <ashutak@gridgain.com>
Committed: Fri Jul 24 16:46:09 2015 +0300

----------------------------------------------------------------------
 .../processors/rest/GridRestProcessor.java      | 154 +++++++++++--------
 1 file changed, 86 insertions(+), 68 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cb51025f/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
index 99e3a8a..d4e84db 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
@@ -83,10 +83,10 @@ public class GridRestProcessor extends GridProcessorAdapter {
     private final LongAdder8 workersCnt = new LongAdder8();
 
     /** SecurityContext map. */
-    private ConcurrentMap<UUID, Session> clientId2Ses = new ConcurrentHashMap<>();
+    private final ConcurrentMap<UUID, UUID> clientId2SesId = new ConcurrentHashMap<>();
 
     /** SecurityContext map. */
-    private ConcurrentMap<UUID, Session> sesTokId2Ses = new ConcurrentHashMap<>();
+    private final ConcurrentMap<UUID, Session> sesId2Ses = new ConcurrentHashMap<>();
 
     /** Protocol handler. */
     private final GridRestProtocolHandler protoHnd = new GridRestProtocolHandler() {
@@ -100,7 +100,7 @@ public class GridRestProcessor extends GridProcessorAdapter {
     };
 
     /** Sesion timeout size */
-    private final long sesTimeout;
+    private final long sesTtl;
 
     /**
      * @param req Request.
@@ -274,78 +274,82 @@ public class GridRestProcessor extends GridProcessorAdapter {
         final UUID clientId = req.clientId();
         final byte[] sesTok = req.sessionToken();
 
-        if (F.isEmpty(sesTok) && clientId == null) {
-            Session ses = new Session();
-            ses.clientId = UUID.randomUUID();
-            ses.sesTokId = UUID.randomUUID();
+        for (int i = 0; i < 10_000; i++) {
+            if (F.isEmpty(sesTok) && clientId == null) {
+                Session ses = Session.random();
 
-            clientId2Ses.put(ses.clientId, ses);
-            sesTokId2Ses.put(ses.sesTokId, ses);
+                if (clientId2SesId.putIfAbsent(clientId, ses.sesId)!=null)
+                    continue; /** New random clientId equals to existing clientId */
 
-            return ses;
-        }
+                sesId2Ses.put(ses.sesId, ses);
 
-        if (F.isEmpty(sesTok) && clientId != null) {
-            Session ses = clientId2Ses.get(clientId);
+                return ses;
+            }
 
-            if (ses == null) { /** First request with this clientId */
-                ses = new Session();
-                ses.clientId = clientId;
-                ses.sesTokId = UUID.randomUUID();
+            if (F.isEmpty(sesTok) && clientId != null) {
+                UUID sesId = clientId2SesId.get(clientId);
 
-                Session curSes = clientId2Ses.putIfAbsent(ses.clientId, ses);
+                if (sesId == null) {
+                    Session ses = Session.fromClientId(clientId);
 
-                if (curSes == null)
-                    sesTokId2Ses.put(ses.sesTokId, ses);
-                else {
-                    boolean expired = curSes.checkTimeoutAndTryUpdateLastTouchTime();
+                    if (clientId2SesId.putIfAbsent(ses.clientId, ses.sesId)!=null)
+                        continue; /** Another thread already register session with the clientId.
*/
+
+                    sesId2Ses.put(ses.sesId, ses);
 
-                    // curSes != null means that there was at least 2 parallel request
-                    // by the same clientId (and we was not first).
-                    // So it was approximately at the same time and oldSession can't be expired.
-                    assert !expired;
+                    return ses;
                 }
+                else {
+                    Session ses = sesId2Ses.get(sesId);
+
+                    if (ses == null)
+                        continue; /** Need to wait while timeout thread complete removing
of timed out sessions. */
 
-                return curSes == null ? ses : curSes;
+                    if (ses.checkTimeoutAndTryUpdateLastTouchTime())
+                        continue; /** Need to wait while timeout thread complete removing
of timed out sessions. */
+
+                    return ses;
+                }
             }
-            else {
+
+            if (!F.isEmpty(sesTok) && clientId == null) {
+                UUID sesId = U.bytesToUuid(sesTok, 0);
+
+                Session ses = sesId2Ses.get(sesId);
+
+                if (ses == null)
+                    throw new IgniteCheckedException("Failed to handle request. Unknown session
token " +
+                        "(maybe expired session). [sessionToken=" + U.byteArray2HexString(sesTok)
+ "]");
+
                 if (ses.checkTimeoutAndTryUpdateLastTouchTime())
-                    return session(req);
+                    continue; /** Need to wait while timeout thread complete removing of
timed out sessions. */
 
                 return ses;
             }
-        }
 
-        if (!F.isEmpty(sesTok) && clientId == null) {
-            UUID sesTokId = U.bytesToUuid(sesTok, 0);
+            if (!F.isEmpty(sesTok) && clientId != null) {
+                UUID sesId = clientId2SesId.get(clientId);
 
-            Session ses = sesTokId2Ses.get(sesTokId);
+                if (!sesId.equals(U.bytesToUuid(sesTok, 0)))
+                    throw new IgniteCheckedException("Failed to handle request. " +
+                        "Unsupported case (misamatched clientId and session token)");
 
-            if (ses == null)
-                throw new IgniteCheckedException("Failed to handle request. Unknown session
token " +
-                    "(maybe expired session). [sessionToken=" + U.byteArray2HexString(sesTok)
+ "]");
+                Session ses = sesId2Ses.get(sesId);
 
-            if (ses.checkTimeoutAndTryUpdateLastTouchTime())
-                return session(req);
+                if (ses == null)
+                    throw new IgniteCheckedException("Failed to handle request. Unknown session
token " +
+                        "(maybe expired session). [sessionToken=" + U.byteArray2HexString(sesTok)
+ "]");
 
-            return ses;
-        }
-
-        if (!F.isEmpty(sesTok) && clientId != null) {
-            Session ses1 = sesTokId2Ses.get(U.bytesToUuid(sesTok, 0));
-            Session ses2 = clientId2Ses.get(clientId);
-
-            if (ses1 == null || ses2 == null || !ses1.equals(ses2))
-                throw new IgniteCheckedException("Failed to handle request. " +
-                    "Unsupported case (use one: clientId or session token)");
+                if (ses.checkTimeoutAndTryUpdateLastTouchTime())
+                    continue; /** Lets try to reslove session again. */
 
-            if (ses1.checkTimeoutAndTryUpdateLastTouchTime())
-                return session(req);
+                return ses;
+            }
 
-            return ses1;
+            throw new IgniteCheckedException("Failed to handle request (Unreachable state).");
         }
 
-        throw new IgniteCheckedException("Failed to handle request (Unreachable state).");
+        throw new IgniteCheckedException("Failed to handle request (Could not resolve session).");
     }
 
     /**
@@ -368,7 +372,7 @@ public class GridRestProcessor extends GridProcessorAdapter {
             sesExpTime0 = DEFAULT_SES_TIMEOUT;
         }
 
-        sesTimeout = sesExpTime0;
+        sesTtl = sesExpTime0;
     }
 
     /** {@inheritDoc} */
@@ -414,20 +418,17 @@ public class GridRestProcessor extends GridProcessorAdapter {
                     while(!Thread.currentThread().isInterrupted()) {
                         Thread.sleep(SES_TIMEOUT_CHECK_DELAY);
 
-                        for (Iterator<Map.Entry<UUID, Session>> iter = clientId2Ses.entrySet().iterator();
+                        for (Iterator<Map.Entry<UUID, Session>> iter = sesId2Ses.entrySet().iterator();
                             iter.hasNext();) {
                             Map.Entry<UUID, Session> e = iter.next();
 
-                            if (e.getValue().checkTimeout(sesTimeout))
-                                iter.remove();
-                        }
+                            Session ses = e.getValue();
 
-                        for (Iterator<Map.Entry<UUID, Session>> iter = sesTokId2Ses.entrySet().iterator();
-                            iter.hasNext();) {
-                            Map.Entry<UUID, Session> e = iter.next();
-
-                            if (e.getValue().isTimedOut())
+                            if (ses.checkTimeout(sesTtl)) {
                                 iter.remove();
+
+                                clientId2SesId.remove(ses.clientId, ses.sesId);
+                            }
                         }
                     }
                 }
@@ -816,10 +817,10 @@ public class GridRestProcessor extends GridProcessorAdapter {
         private static final Long TIMEDOUT_FLAG = 0L;
 
         /** Client id. */
-        private volatile UUID clientId;
+        private final UUID clientId;
 
         /** Session token id. */
-        private volatile UUID sesTokId;
+        private final UUID sesId;
 
         /** Security context. */
         private volatile SecurityContext secCtx;
@@ -830,11 +831,28 @@ public class GridRestProcessor extends GridProcessorAdapter {
          */
         private final AtomicLong lastTouchTime = new AtomicLong(U.currentTimeMillis());
 
+        private Session(UUID clientId, UUID sesId) {
+            this.clientId = clientId;
+            this.sesId = sesId;
+        }
+
+        static Session random() {
+            return new Session(UUID.randomUUID(), UUID.randomUUID());
+        }
+
+        static Session fromClientId(UUID clientId) {
+            return new Session(clientId, UUID.randomUUID());
+        }
+
+        static Session fromSessionToken(UUID sesTokId) {
+            return new Session(UUID.randomUUID(), sesTokId);
+        }
+
         /**
          * @return Session token as bytes.
          */
         byte[] sesTok(){
-            return U.uuidToBytes(sesTokId);
+            return U.uuidToBytes(sesId);
         }
 
         /**
@@ -855,7 +873,7 @@ public class GridRestProcessor extends GridProcessorAdapter {
         /**
          * Checks whether session at expired state (EPIRATION_FLAG) or not, if not then tries
to update last touch time.
          *
-         * @return <code>True</code> if expired.
+         * @return <code>True</code> if timed out.
          */
         boolean checkTimeoutAndTryUpdateLastTouchTime() {
             while (true) {
@@ -889,7 +907,7 @@ public class GridRestProcessor extends GridProcessorAdapter {
 
             if (clientId != null ? !clientId.equals(ses.clientId) : ses.clientId != null)
                 return false;
-            if (sesTokId != null ? !sesTokId.equals(ses.sesTokId) : ses.sesTokId != null)
+            if (sesId != null ? !sesId.equals(ses.sesId) : ses.sesId != null)
                 return false;
 
             return true;
@@ -898,7 +916,7 @@ public class GridRestProcessor extends GridProcessorAdapter {
         /** {@inheritDoc} */
         @Override public int hashCode() {
             int res = clientId != null ? clientId.hashCode() : 0;
-            res = 31 * res + (sesTokId != null ? sesTokId.hashCode() : 0);
+            res = 31 * res + (sesId != null ? sesId.hashCode() : 0);
             return res;
         }
     }


Mime
View raw message