org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer クラス

org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer の継承関係図

org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer 連携図

クラス
class	WorkerResult

公開メンバ関数
	InfinispanCacheInitializer (KeycloakSessionFactory sessionFactory, Cache< String, Serializable > workCache, SessionLoader sessionLoader, String stateKeySuffix, int sessionsPerSegment, int maxErrors)
void	initCache ()
Cache< String, Serializable >	getWorkCache ()
void	loadSessions ()

限定公開メンバ関数
void	startLoading ()
void	startLoadingImpl (InitializerState state, SessionLoader.LoaderContext ctx)
boolean	isFinished ()
boolean	isCoordinator ()
InitializerState	getStateFromCache ()
void	saveStateToCache (final InitializerState state)

限定公開変数類
final KeycloakSessionFactory	sessionFactory
final Cache< String, Serializable >	workCache
final SessionLoader	sessionLoader
final int	sessionsPerSegment
final String	stateKey

非公開変数類
final int	maxErrors

静的非公開変数類
static final Logger	log = Logger.getLogger(InfinispanCacheInitializer.class)

詳解

Startup initialization for reading persistent userSessions to be filled into infinispan/memory . In cluster, the initialization is distributed among all cluster nodes, so the startup time is even faster

Implementation is pretty generic and doesn't contain any "userSession" specific stuff. All logic related to how are sessions loaded is in the SessionLoader implementation

著者

Marek Posolda

構築子と解体子

InfinispanCacheInitializer()

org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer.InfinispanCacheInitializer ( KeycloakSessionFactory  sessionFactory, Cache< String, Serializable >  workCache, SessionLoader  sessionLoader, String  stateKeySuffix, int  sessionsPerSegment, int  maxErrors  )

inline

                                                                                                                                                                                                               {
        super(sessionFactory, workCache, sessionLoader, stateKeySuffix, sessionsPerSegment);
        this.maxErrors = maxErrors;
    }

関数詳解

getStateFromCache()

InitializerState org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.getStateFromCache ( )

inlineprotectedinherited

                                                   {
        // We ignore cacheStore for now, so that in Cross-DC scenario (with RemoteStore enabled) is the remoteStore ignored.
        return (InitializerState) workCache.getAdvancedCache()
                .withFlags(Flag.SKIP_CACHE_STORE, Flag.SKIP_CACHE_LOAD)
                .get(stateKey);
    }

getWorkCache()

Cache<String, Serializable> org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.getWorkCache ( )

inlineinherited

                                                      {
        return workCache;
    }

initCache()

void org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer.initCache ( )

inline

                            {
        this.workCache.getAdvancedCache().getComponentRegistry().registerComponent(sessionFactory, KeycloakSessionFactory.class);
    }

isCoordinator()

boolean org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.isCoordinator ( )

inlineprotectedinherited

                                      {
        Transport transport = workCache.getCacheManager().getTransport();
        return transport == null || transport.isCoordinator();
    }

isFinished()

boolean org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.isFinished ( )

inlineprotectedinherited

                                   {
        // Check if we should skipLoadingSessions. This can happen if someone else already did the task (For example in cross-dc environment, it was done by different DC)
        boolean isFinishedAlready = this.sessionLoader.isFinished(this);
        if (isFinishedAlready) {
            return true;
        }

        InitializerState state = getStateFromCache();
        return state != null && state.isFinished();
    }

loadSessions()

void org.keycloak.models.sessions.infinispan.initializer.CacheInitializer.loadSessions ( )

inlineinherited

                               {
        while (!isFinished()) {
            if (!isCoordinator()) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                    log.error("Interrupted", ie);
                }
            } else {
                startLoading();
            }
        }
    }

saveStateToCache()

void org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.saveStateToCache ( final InitializerState  state )

inlineprotectedinherited

                                                                  {

        // 3 attempts to send the message (it may fail if some node fails in the meantime)
        retry(3, new Runnable() {

            @Override
            public void run() {

                // Save this synchronously to ensure all nodes read correct state
                // We ignore cacheStore for now, so that in Cross-DC scenario (with RemoteStore enabled) is the remoteStore ignored.
                BaseCacheInitializer.this.workCache.getAdvancedCache().
                        withFlags(Flag.IGNORE_RETURN_VALUES, Flag.FORCE_SYNCHRONOUS, Flag.SKIP_CACHE_STORE, Flag.SKIP_CACHE_LOAD)
                        .put(stateKey, state);
            }

        });
    }

startLoading()

void org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer.startLoading ( )

inlineprotected

                                  {
        InitializerState state = getStateFromCache();
        SessionLoader.LoaderContext[] ctx = new SessionLoader.LoaderContext[1];
        if (state == null) {
            // Rather use separate transactions for update and counting
            KeycloakModelUtils.runJobInTransaction(sessionFactory, new KeycloakSessionTask() {
                @Override
                public void run(KeycloakSession session) {
                    sessionLoader.init(session);
                }

            });

            KeycloakModelUtils.runJobInTransaction(sessionFactory, new KeycloakSessionTask() {
                @Override
                public void run(KeycloakSession session) {
                    ctx[0] = sessionLoader.computeLoaderContext(session);
                }

            });

            state = new InitializerState(ctx[0].getSegmentsCount());
            saveStateToCache(state);
        } else {
            KeycloakModelUtils.runJobInTransaction(sessionFactory, new KeycloakSessionTask() {
                @Override
                public void run(KeycloakSession session) {
                    ctx[0] = sessionLoader.computeLoaderContext(session);
                }

            });
        }

        log.debugf("Start loading with loader: '%s', ctx: '%s' , state: %s",
                sessionLoader.toString(), ctx[0].toString(), state.toString());

        startLoadingImpl(state, ctx[0]);
    }

startLoadingImpl()

void org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer.startLoadingImpl ( InitializerState  state, SessionLoader.LoaderContext  ctx  )

inlineprotected

                                                                                             {
        // Assume each worker has same processor's count
        int processors = Runtime.getRuntime().availableProcessors();

        ExecutorService localExecutor = Executors.newCachedThreadPool();
        Transport transport = workCache.getCacheManager().getTransport();
        boolean distributed = transport != null;
        ExecutorService executorService = distributed ? new DefaultExecutorService(workCache, localExecutor) : localExecutor;

        int errors = 0;

        try {
            while (!state.isFinished()) {
                int nodesCount = transport==null ? 1 : transport.getMembers().size();
                int distributedWorkersCount = processors * nodesCount;

                log.debugf("Starting next iteration with %d workers", distributedWorkersCount);

                List<Integer> segments = state.getUnfinishedSegments(distributedWorkersCount);

                if (log.isTraceEnabled()) {
                    log.trace("unfinished segments for this iteration: " + segments);
                }

                List<Future<WorkerResult>> futures = new LinkedList<>();
                for (Integer segment : segments) {
                    SessionInitializerWorker worker = new SessionInitializerWorker();
                    worker.setWorkerEnvironment(segment, ctx, sessionLoader);
                    if (!distributed) {
                        worker.setEnvironment(workCache, null);
                    }

                    Future<WorkerResult> future = executorService.submit(worker);
                    futures.add(future);
                }

                for (Future<WorkerResult> future : futures) {
                    try {
                        WorkerResult result = future.get();

                        if (result.getSuccess()) {
                            int computedSegment = result.getSegment();
                            state.markSegmentFinished(computedSegment);
                        } else {
                            if (log.isTraceEnabled()) {
                                log.tracef("Segment %d failed to compute", result.getSegment());
                            }
                        }
                    } catch (InterruptedException ie) {
                        errors++;
                        log.error("Interruped exception when computed future. Errors: " + errors, ie);
                    } catch (ExecutionException ee) {
                        errors++;
                        log.error("ExecutionException when computed future. Errors: " + errors, ee);
                    }
                }

                if (errors >= maxErrors) {
                    throw new RuntimeException("Maximum count of worker errors occured. Limit was " + maxErrors + ". See server.log for details");
                }

                saveStateToCache(state);

                log.debugf("New initializer state pushed. The state is: %s", state);
            }

            // Loader callback after the task is finished
            this.sessionLoader.afterAllSessionsLoaded(this);

        } finally {
            if (distributed) {
                executorService.shutdown();
            }
            localExecutor.shutdown();
        }
    }

メンバ詳解

log

final Logger org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer.log = Logger.getLogger(InfinispanCacheInitializer.class)

staticprivate

maxErrors

final int org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitializer.maxErrors

private

sessionFactory

final KeycloakSessionFactory org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.sessionFactory

protectedinherited

sessionLoader

final SessionLoader org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.sessionLoader

protectedinherited

sessionsPerSegment

final int org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.sessionsPerSegment

protectedinherited

stateKey

final String org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.stateKey

protectedinherited

workCache

final Cache<String, Serializable> org.keycloak.models.sessions.infinispan.initializer.BaseCacheInitializer.workCache

protectedinherited

---

このクラス詳解は次のファイルから抽出されました:

• D:/AppData/doxygen/keycloak/src/keycloak/src/main/java/org/keycloak/models/sessions/infinispan/initializer/InfinispanCacheInitializer.java