From e958f966da1eeaf11d346ad9326469b87fd9689f Mon Sep 17 00:00:00 2001 From: Metin Dumandag Date: Fri, 21 May 2021 10:47:16 +0300 Subject: [PATCH] [API-424] Do not allow concurrent requests on SessionManager for the same group id (#904) * Do not allow concurrent requests on SessionManager for the same group id We should make sure that for the same group id, new create session requests should only be made if there is no session or the session is expired. To achieve this, a new Map that holds the group id to in-flight create session requests pairs is introduced. When an in-flight request is found for a group id, concurrent requests now wait for the initial request to complete (either normally or exceptionally) and then try again. * check the error text in the test --- src/proxy/cpsubsystem/CPSessionManager.ts | 44 ++++++++++---- .../proxy/cpsubsystem/CPSessionManagerTest.js | 60 +++++++++++++++++-- 2 files changed, 89 insertions(+), 15 deletions(-) diff --git a/src/proxy/cpsubsystem/CPSessionManager.ts b/src/proxy/cpsubsystem/CPSessionManager.ts index ca7777bcf..930ffac23 100644 --- a/src/proxy/cpsubsystem/CPSessionManager.ts +++ b/src/proxy/cpsubsystem/CPSessionManager.ts @@ -87,6 +87,8 @@ export class CPSessionManager { // map private readonly sessions: Map = new Map(); + // group id to in-flight create session requests map + private readonly inFlightCreateSessionRequests: Map> = new Map(); private heartbeatTask: Task; private isShutdown = false; @@ -142,7 +144,10 @@ export class CPSessionManager { .catch((e) => { this.logger.debug('CPSessionManager', 'Could not close CP sessions.', e); }) - .then(() => this.sessions.clear()); + .then(() => { + this.sessions.clear(); + this.inFlightCreateSessionRequests.clear(); + }); } private getOrCreateSession(groupId: RaftGroupId): Promise { @@ -157,21 +162,38 @@ export class CPSessionManager { } private createNewSession(groupId: RaftGroupId): Promise { - return this.requestNewSession(groupId).then((response) => { - const state = new SessionState(response.sessionId, groupId, response.ttlMillis.toNumber()); - this.sessions.set(groupId.getStringId(), state); - this.scheduleHeartbeatTask(response.heartbeatMillis.toNumber()); - return state; - }); + // Check if there is a session request for this group id in-flight. + const inFlightRequest = this.inFlightCreateSessionRequests.get(groupId.getStringId()); + if (inFlightRequest === undefined) { + // No in-flight request for this group id. Let's make a new one and register it. + const requestNewSessionPromise = this.requestNewSession(groupId).then((response) => { + const state = new SessionState(response.sessionId, groupId, response.ttlMillis.toNumber()); + this.sessions.set(groupId.getStringId(), state); + this.scheduleHeartbeatTask(response.heartbeatMillis.toNumber()); + return state; + }); + this.inFlightCreateSessionRequests.set(groupId.getStringId(), requestNewSessionPromise); + + // Remove the request once it is completed normally or exceptionally. + // This ensures that, if there is no session(request failed or session somehow removed later) + // or the session is expired, later, a new request can be made for the same group id. + const onResponseOrError = () => this.inFlightCreateSessionRequests.delete(groupId.getStringId()); + requestNewSessionPromise.then(onResponseOrError, onResponseOrError); + return requestNewSessionPromise; + } else { + // There is an in-flight session request for this group id. We should + // wait for it to complete normally or exceptionally, and then try again. + // There should be no concurrent create session requests for the same + // group id. + const onResponseOrError = () => this.getOrCreateSession(groupId); + return inFlightRequest.then(onResponseOrError, onResponseOrError); + } } private requestNewSession(groupId: RaftGroupId): Promise { const clientMessage = CPSessionCreateSessionCodec.encodeRequest(groupId, this.clientName); return this.invocationService.invokeOnRandomTarget(clientMessage) - .then((clientMessage) => { - const response = CPSessionCreateSessionCodec.decodeResponse(clientMessage); - return response; - }); + .then(CPSessionCreateSessionCodec.decodeResponse); } private requestCloseSession(groupId: RaftGroupId, sessionId: Long): Promise { diff --git a/test/unit/proxy/cpsubsystem/CPSessionManagerTest.js b/test/unit/proxy/cpsubsystem/CPSessionManagerTest.js index 8c1fb5bfb..5d45b2181 100644 --- a/test/unit/proxy/cpsubsystem/CPSessionManagerTest.js +++ b/test/unit/proxy/cpsubsystem/CPSessionManagerTest.js @@ -32,6 +32,7 @@ const { } = require('../../../../lib/proxy/cpsubsystem/CPSessionManager'); const { DefaultLogger } = require('../../../../lib/logging/DefaultLogger'); const { RaftGroupId } = require('../../../../lib/proxy/cpsubsystem/RaftGroupId'); +const { deferredPromise } = require('../../../../lib/util/Util'); describe('CPSessionManagerTest', function () { @@ -114,13 +115,17 @@ describe('CPSessionManagerTest', function () { return new SessionState(Long.fromNumber(SESSION_ID), groupId, TTL_MILLIS); } - function stubRequestNewSession() { - const stub = sandbox.stub(sessionManager, 'requestNewSession'); - stub.returns(Promise.resolve({ + function prepareNewSessionResponse() { + return { sessionId: Long.fromNumber(SESSION_ID), ttlMillis: Long.fromNumber(TTL_MILLIS), heartbeatMillis: Long.fromNumber(HEARTBEAT_MILLIS) - })); + }; + } + + function stubRequestNewSession() { + const stub = sandbox.stub(sessionManager, 'requestNewSession'); + stub.returns(Promise.resolve(prepareNewSessionResponse())); return stub; } @@ -200,6 +205,53 @@ describe('CPSessionManagerTest', function () { expect(sessionManager.sessions.get(GROUP_ID_AS_STRING).acquireCount).to.be.equal(1); }); + it('acquireSession: should not request new session for the same group id for concurrent requests', async function () { + const stub = sandbox.stub(sessionManager, 'requestNewSession'); + const deferred = deferredPromise(); + stub.returns(deferred.promise); + + const groupId = prepareGroupId(); + + const acquireSessionPromises = [ + sessionManager.acquireSession(groupId), + sessionManager.acquireSession(groupId), + sessionManager.acquireSession(groupId), + ]; + + expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(1); + + deferred.resolve(prepareNewSessionResponse()); + + await Promise.all(acquireSessionPromises); + expect(stub.withArgs(groupId).calledOnce).to.be.true; + expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(0); + }); + + it('acquireSession: should request new sessions for concurrent requests when requests fail', async function () { + const stub = sandbox.stub(sessionManager, 'requestNewSession'); + const deferred = deferredPromise(); + stub.returns(deferred.promise); + + const groupId = prepareGroupId(); + + const acquireSessionPromises = [ + sessionManager.acquireSession(groupId), + sessionManager.acquireSession(groupId), + sessionManager.acquireSession(groupId), + ]; + + expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(1); + + deferred.resolve(Promise.reject(new Error('expected'))); + + for (const acquireSessionPromise of acquireSessionPromises) { + await expect(acquireSessionPromise).to.be.rejectedWith(Error, 'expected'); + } + + expect(stub.withArgs(groupId).callCount).to.be.equal(acquireSessionPromises.length); + expect(sessionManager.inFlightCreateSessionRequests.size).to.be.equal(0); + }); + it('releaseSession: should decrement acquire counter by 1 for known session', function () { sessionManager.sessions.set(GROUP_ID_AS_STRING, prepareSessionState());