[Android/Java] Glide, source analysis. -- 3. EngineJob & DecodeJob

Glide

Engine Job & Decode Job

지난 포스팅에서는...

2016/11/02 - [프로그래밍/Android] - [Android/Java] Glide, An Image Loader. -- 1. Overview

2016/12/09 - [프로그래밍/Android] - [Android/Java] Glide, source analysis. -- 2. Flow

 - Glide에대한 개괄적인 소개와 with(), load(), into()의 flow에 대해서 알아보았다.

   다음으로는 조금 더 심도있게 접근하여 어째서 Glide가 빠르고 안정적으로 동작할 수 있는가에 대해서 알아보도록 한다.

   그러기 위해서 Glide에 대한 포스팅을 잠시 중단하고 병렬처리에 관한 포스팅을 3부작으로 올렸었다. (내가 연재를 쉰것은 추진력을 얻기 위함이었다)

2016/12/01 - [프로그래밍/Android] - [Android/Java] 병렬 프로그래밍 : Executor Framework에대한 고찰 ----- 1

2016/12/05 - [프로그래밍/Android] - [Android/Java] 병렬 프로그래밍 : Executor Framework에대한 고찰 ----- 2

2016/12/07 - [프로그래밍/Android] - [Android/Java] 병렬 프로그래밍 : Executor Framework에대한 고찰 ----- 3

2016/12/06 - [프로그래밍/Android] - [Android/Java] 변수를 Volatile로 선언하면?

Engine Job.

 - "MemoryCache, DiskCache 그리고 특정 위치중 어느곳에서 가져올 것인가?" 에 대한 작업을 병렬로 처리한다.

 - Glide의 핵심작업 영역을 알아보기 위해서는 SingleRequest.java의 onSizeReady()콜백을 받은 이후부터 살펴본다.

   

// SingleRequest.java
 
/**
 * A callback method that should never be invoked directly.
 */
@Override
public void onSizeReady(int width, int height) {
  stateVerifier.throwIfRecycled();
  if (Log.isLoggable(TAG, Log.VERBOSE)) {
    logV("Got onSizeReady in " + LogTime.getElapsedMillis(startTime));
  }
  if (status != Status.WAITING_FOR_SIZE) {
    return;
  }
  status = Status.RUNNING;
 
  float sizeMultiplier = requestOptions.getSizeMultiplier();
  this.width = maybeApplySizeMultiplier(width, sizeMultiplier);
  this.height = maybeApplySizeMultiplier(height, sizeMultiplier);
 
  if (Log.isLoggable(TAG, Log.VERBOSE)) {
    logV("finished setup for calling load in " + LogTime.getElapsedMillis(startTime));
  }
  loadStatus = engine.load(
      glideContext,
      model,
      requestOptions.getSignature(),
      this.width,
      this.height,
      requestOptions.getResourceClass(),
      transcodeClass,
      priority,
      requestOptions.getDiskCacheStrategy(),
      requestOptions.getTransformations(),
      requestOptions.isTransformationRequired(),
      requestOptions.getOptions(),
      requestOptions.isMemoryCacheable(),
      requestOptions.getUseUnlimitedSourceGeneratorsPool(),
      this);
  if (Log.isLoggable(TAG, Log.VERBOSE)) {
    logV("finished onSizeReady in " + LogTime.getElapsedMillis(startTime));
  }
}

    - engine.load()를 호출하는 것 말고는 특이한 것은 없다.  Glide는 size에 맞게 리소스를 가져오기 때문에 onSizeReady()콜백 이후부터가 가져오는 부분이 된다.

<<Engine.load() flow>>

[이해를 돕기위해 추가한 점선(---)은 데이터 플로우를 나타낸다]

1. 가져오려는 리소스가 캐쉬에 있는지 확인한다.
2. 가져오려는 리소스가 액티브인지 확인한다.
• 액티브-리소스는 하나 이상의 request에 제공되었지만 아직 릴리스되지 않은 리소스이다. 
• 리소스의 모든 소비자가 해당 리소스를 해제하면 리소스가 캐시된다.
• 리소스가 캐시에서 새 소비자(consumer)에게로 반환되면 액티브-리소스에 다시 추가된다.
• 리소스가 캐시에서 제거되면 가능한 경우 리소스가 재활용되고 다시 사용되며 리소스가 삭제된다.
3. 가져오려는 리소스가 이미 작업중인지 확인한다. 
4. 새 작업을 만든다.
• load()이 호출될 때 ResourceCallback을 받았고, 이것을 새 작업에도 넘겨준다.
• 이후 Decode가 완료되면 onResourceReady() 콜백을 준다.

• 
  

    public <R> LoadStatus load(
        GlideContext glideContext,
        Object model,
        Key signature,
        int width,
        int height,
        Class<?> resourceClass,
        Class<R> transcodeClass,
        Priority priority,
        DiskCacheStrategy diskCacheStrategy,
        Map<Class<?>, Transformation<?>> transformations,
        boolean isTransformationRequired,
        Options options,
        boolean isMemoryCacheable,
        boolean useUnlimitedSourceExecutorPool,
        ResourceCallback cb) {
      Util.assertMainThread();
      long startTime = LogTime.getLogTime();
   
      EngineKey key = keyFactory.buildKey(model, signature, width, height, transformations,
          resourceClass, transcodeClass, options);
   
      EngineResource<?> cached = loadFromCache(key, isMemoryCacheable);
      if (cached != null) {
        cb.onResourceReady(cached, DataSource.MEMORY_CACHE);
        if (Log.isLoggable(TAG, Log.VERBOSE)) {
          logWithTimeAndKey("Loaded resource from cache", startTime, key);
        }
        return null;
      }
   
      EngineResource<?> active = loadFromActiveResources(key, isMemoryCacheable);
      if (active != null) {
        cb.onResourceReady(active, DataSource.MEMORY_CACHE);
        if (Log.isLoggable(TAG, Log.VERBOSE)) {
          logWithTimeAndKey("Loaded resource from active resources", startTime, key);
        }
        return null;
      }
   
      EngineJob<?> current = jobs.get(key);
      if (current != null) {
        current.addCallback(cb);
        if (Log.isLoggable(TAG, Log.VERBOSE)) {
          logWithTimeAndKey("Added to existing load", startTime, key);
        }
        return new LoadStatus(cb, current);
      }
   
      EngineJob<R> engineJob = engineJobFactory.build(key, isMemoryCacheable,
          useUnlimitedSourceExecutorPool);
      DecodeJob<R> decodeJob = decodeJobFactory.build(
          glideContext,
          model,
          key,
          signature,
          width,
          height,
          resourceClass,
          transcodeClass,
          priority,
          diskCacheStrategy,
          transformations,
          isTransformationRequired,
          options,
          engineJob);
      jobs.put(key, engineJob);
      engineJob.addCallback(cb);
      engineJob.start(decodeJob);
   
      if (Log.isLoggable(TAG, Log.VERBOSE)) {
        logWithTimeAndKey("Started new load", startTime, key);
      }
      return new LoadStatus(cb, engineJob);
    }

<<EngineJob 만들기>>

// Engine.java
 
static class EngineJobFactory {
  @Synthetic final GlideExecutor diskCacheExecutor;
  @Synthetic final GlideExecutor sourceExecutor;
  @Synthetic final GlideExecutor sourceUnlimitedExecutor;
  @Synthetic final EngineJobListener listener;
  @Synthetic final Pools.Pool<EngineJob<?>> pool = FactoryPools.simple(JOB_POOL_SIZE,
      new FactoryPools.Factory<EngineJob<?>>() {
        @Override
        public EngineJob<?> create() {
          return new EngineJob<Object>(diskCacheExecutor, sourceExecutor, sourceUnlimitedExecutor,
              listener, pool);
        }
      });
 
  EngineJobFactory(GlideExecutor diskCacheExecutor, GlideExecutor sourceExecutor,
      GlideExecutor sourceUnlimitedExecutor, EngineJobListener listener) {
    this.diskCacheExecutor = diskCacheExecutor;
    this.sourceExecutor = sourceExecutor;
    this.sourceUnlimitedExecutor = sourceUnlimitedExecutor;
    this.listener = listener;
  }
 
  @SuppressWarnings("unchecked")
  <R> EngineJob<R> build(Key key, boolean isMemoryCacheable,
      boolean useUnlimitedSourceGeneratorPool) {
    EngineJob<R> result = (EngineJob<R>) pool.acquire();
    return result.init(key, isMemoryCacheable, useUnlimitedSourceGeneratorPool);
  }
}

 - Engine을 만들 때에 사용하는 EngineJobFactory이다.

   사전지식용으로 포스팅했던 Java 병렬처리 시리즈 ( Executor Framework에대한 고찰)는 제목 그대로 Java(특히 6 이후)에 이미 있는 Executor, 특히 마지막에는 ThreadPoolExecutor를 사용했었으나, 어떤 이유인지 EngineJob에는 그것을 사용하지 않고 직접 Pool과 map을 만들어 관리한다. 
   이상한것은 EngineJob이 들고있는 MemoryCacheExecutor, SourceExecutor, DiskCacheExecutor등 GlideExecutor는 ThreadPoolExecutor를 상속받아 구현되었다.

   왜 이렇게 되었는지도 한번 확인해보자.

• EngineJobFactory는 EngineJob을 관리하는 pool을 SimplePool로 직접 생성하여 가지고있다. (SimplePool은 ThreadSafe하지 않은데, 어떻게 관리하였는지 나중에 살펴본다.)
• pool은 150개(JOB_POOL_SIZE)의 size를 가진다.
• build() 메소드를 보면 pool에서 EngineJob을 가져오는데, pool은 사실 SimplePool이 아니라 SimplePool을 생성/관리하는 FactoryPool이다.
• 따라서 28번행의 pool.acquire();를 하면, FactoryPool의 acquire()이 호출되고, 이것은 null인경우 pool 하나를 생성하여 반환한다.
  

// FactoryPools.java
 
private static final class FactoryPool<T> implements Pool<T> {
  private final Factory<T> factory;
  private final Resetter<T> resetter;
  private final Pool<T> pool;
 
  FactoryPool(Pool<T> pool, Factory<T> factory, Resetter<T> resetter) {
    this.pool = pool;
    this.factory = factory;
    this.resetter = resetter;
  }
 
  @Override
  public T acquire() {
    T result = pool.acquire();
    if (result == null) {
      result = factory.create();
      if (Log.isLoggable(TAG, Log.VERBOSE)) {
        Log.v(TAG, "Created new " + result.getClass());
      }
    }
    if (result instanceof Poolable) {
      ((Poolable) result).getVerifier().setRecycled(false /*isRecycled*/);
    }
    return result;
  }
 
  @Override
  public boolean release(T instance) {
    if (instance instanceof Poolable) {
      ((Poolable) instance).getVerifier().setRecycled(true /*isRecycled*/);
    }
    resetter.reset(instance);
    return pool.release(instance);
  }
}

[SimplePool에서 acquire()하지 못한 경우 SimplePool을 하나 만든다]

  - FactoryPool의 구현을 통해

1. 절대로 null을 리턴하지 않는다
2. 생성시 로그를 찍는다
3. pool 내에있는 동안 객체가 사용되지 않는것을 보장한다.

<<Engine - multi thread 구성>>

1. EngineJobFactory를 이용하여 EngineJob pool에서 하나 가져온다. (만약 없으면 생성)
   
2. Engine도 생성된 EngineJob을 map에 들고있는다. (왜? 아마도 cancel()처리를 하기 위해서인 것 같은데 future를 관리하는 것 보다는 편한것 같다.)
3. DecodeJobFactory를 이용하여 DecodeJob pool에서 하나 가져온다 (없으면 생성)
4. 생성하여 EngineJob이 가지고있는 GlideExecutor에서 실행한다.
1. Cache에 있는지 확인
2. 없으면 Decode작업
3. 완료되면 onResourceReady() 콜백

Decode Job.

// DecodeJob.java
 
@Override
public void run() {
  // This should be much more fine grained, but since Java's thread pool implementation silently
  // swallows all otherwise fatal exceptions, this will at least make it obvious to developers
  // that something is failing.
  try {
    if (isCancelled) {
      notifyFailed();
      return;
    }
    runWrapped();
  } catch (RuntimeException e) {
    if (Log.isLoggable(TAG, Log.DEBUG)) {
      Log.d(TAG, "DecodeJob threw unexpectedly"
          + ", isCancelled: " + isCancelled
          + ", stage: " + stage, e);
    }
    // When we're encoding we've already notified our callback and it isn't safe to do so again.
    if (stage != Stage.ENCODE) {
      notifyFailed();
    }
    if (!isCancelled) {
      throw e;
    }
  }
}
 
private void runWrapped() {
   switch (runReason) {
    case INITIALIZE:
      stage = getNextStage(Stage.INITIALIZE);
      currentGenerator = getNextGenerator();
      runGenerators();
      break;
    case SWITCH_TO_SOURCE_SERVICE:
      runGenerators();
      break;
    case DECODE_DATA:
      decodeFromRetrievedData();
      break;
    default:
      throw new IllegalStateException("Unrecognized run reason: " + runReason);
  }
}

 - DecodeJob은 ThreadPoolExecutor에 의해 동작해야 하므로 Runnable이던지 Callable이어야 한다. Glide에서는 Runnable로 작성하였다.

 - run()메소드는 조금 더 정교하면 좋겠지만, Java의 thread pool은 fatal exception에대해 그냥 씹어버리므로... 여기서는 최소한 개발자에게만큼은 무언가 잘못되었다는 것을 알릴 수 있도록 구현하였다.

 - runWrapped()에서는 switch()문을 사용하여 어떤 단계를 거쳐가는지 명확히 하였다. 실제로 동작하는 코드는 다음과 같다.

// DecodeJob.java
 
private void runGenerators() {
  currentThread = Thread.currentThread();
  startFetchTime = LogTime.getLogTime();
  boolean isStarted = false;
  while (!isCancelled && currentGenerator != null
      && !(isStarted = currentGenerator.startNext())) {
    stage = getNextStage(stage);
    currentGenerator = getNextGenerator();
 
    if (stage == Stage.SOURCE) {
      reschedule();
      return;
    }
  }
  // We've run out of stages and generators, give up.
  if ((stage == Stage.FINISHED || isCancelled) && !isStarted) {
    notifyFailed();
  }
 
  // Otherwise a generator started a new load and we expect to be called back in
  // onDataFetcherReady.
}

 - 순서도상에 별도 언급은 없었으나, DecodeJob은 ViewTreeObserver에 걸어둔 LifeCycleListener에 의해서 cancel()이 불리기도 한다.

 - 위의 코드에서 재미있는 것은 13번째 행의 reschedule()이다.

1. 데이터 확인.
1. ResourceCahce를 확인한다.
2. DataCache를 확인한다.
3. 1,2번의 캐쉬에 없다면 해당 소스를 새로 가져와야한다는 것인데, 이때에는 ThreadPoolExecutor에 다시 excute()한다. (reschedule()) --> 아마도 캐쉬에 있는것 먼저 처리하려고 하는 듯 하다.
2. 로드

1. 캐쉬에 있는지를 검사한 후에는 해당하는 Loader의 DataFetcher<T>를 이용하여 load한다.
   해당하는 Loader는 Glide객체를 생성할 때 이미 정의해놓았었다. (하기 코드 참조, 모델의 Class에 해당하는 로더를 가져온다.)

   // Glide.java
    
   registry = new Registry()
           .register(ByteBuffer.class, new ByteBufferEncoder())
           .register(InputStream.class, new StreamEncoder(arrayPool))
           /* Bitmaps */
           .append(ByteBuffer.class, Bitmap.class,
               new ByteBufferBitmapDecoder(downsampler))
           .append(InputStream.class, Bitmap.class,
               new StreamBitmapDecoder(downsampler, arrayPool))
           .append(ParcelFileDescriptor.class, Bitmap.class, new VideoBitmapDecoder(bitmapPool))
           .register(Bitmap.class, new BitmapEncoder())
           /* GlideBitmapDrawables */
           .append(ByteBuffer.class, BitmapDrawable.class,
               new BitmapDrawableDecoder<>(resources, bitmapPool,
                   new ByteBufferBitmapDecoder(downsampler)))
           .append(InputStream.class, BitmapDrawable.class,
               new BitmapDrawableDecoder<>(resources, bitmapPool,
                   new StreamBitmapDecoder(downsampler, arrayPool)))
           .append(ParcelFileDescriptor.class, BitmapDrawable.class,
               new BitmapDrawableDecoder<>(resources, bitmapPool, new VideoBitmapDecoder(bitmapPool)))
           .register(BitmapDrawable.class, new BitmapDrawableEncoder(bitmapPool, new BitmapEncoder()))
           /* GIFs */
           .prepend(InputStream.class, GifDrawable.class,
               new StreamGifDecoder(byteBufferGifDecoder, arrayPool))
           .prepend(ByteBuffer.class, GifDrawable.class, byteBufferGifDecoder)
           .register(GifDrawable.class, new GifDrawableEncoder())
           /* GIF Frames */
           .append(GifDecoder.class, GifDecoder.class, new UnitModelLoader.Factory<GifDecoder>())
           .append(GifDecoder.class, Bitmap.class, new GifFrameResourceDecoder(bitmapPool))
           /* Files */
           .register(new ByteBufferRewinder.Factory())
           .append(File.class, ByteBuffer.class, new ByteBufferFileLoader.Factory())
           .append(File.class, InputStream.class, new FileLoader.StreamFactory())
           .append(File.class, File.class, new FileDecoder())
           .append(File.class, ParcelFileDescriptor.class, new FileLoader.FileDescriptorFactory())
           .append(File.class, File.class, new UnitModelLoader.Factory<File>())
           /* Models */
           .register(new InputStreamRewinder.Factory(arrayPool))
           .append(int.class, InputStream.class, new ResourceLoader.StreamFactory(resources))
           .append(
                   int.class,
                   ParcelFileDescriptor.class,
                   new ResourceLoader.FileDescriptorFactory(resources))
           .append(Integer.class, InputStream.class, new ResourceLoader.StreamFactory(resources))
           .append(
                   Integer.class,
                   ParcelFileDescriptor.class,
                   new ResourceLoader.FileDescriptorFactory(resources))
           .append(String.class, InputStream.class, new DataUrlLoader.StreamFactory())
           .append(String.class, InputStream.class, new StringLoader.StreamFactory())
           .append(String.class, ParcelFileDescriptor.class, new StringLoader.FileDescriptorFactory())
           .append(Uri.class, InputStream.class, new HttpUriLoader.Factory())
           .append(Uri.class, InputStream.class, new AssetUriLoader.StreamFactory(context.getAssets()))
           .append(
                   Uri.class,
                   ParcelFileDescriptor.class,
                   new AssetUriLoader.FileDescriptorFactory(context.getAssets()))
           .append(Uri.class, InputStream.class, new MediaStoreImageThumbLoader.Factory(context))
           .append(Uri.class, InputStream.class, new MediaStoreVideoThumbLoader.Factory(context))
           .append(
               Uri.class,
                InputStream.class,
                new UriLoader.StreamFactory(context.getContentResolver()))
           .append(Uri.class, ParcelFileDescriptor.class,
                new UriLoader.FileDescriptorFactory(context.getContentResolver()))
           .append(Uri.class, InputStream.class, new UrlUriLoader.StreamFactory())
           .append(URL.class, InputStream.class, new UrlLoader.StreamFactory())
           .append(Uri.class, File.class, new MediaStoreFileLoader.Factory(context))
           .append(GlideUrl.class, InputStream.class, new HttpGlideUrlLoader.Factory())
           .append(byte[].class, ByteBuffer.class, new ByteArrayLoader.ByteBufferFactory())
           .append(byte[].class, InputStream.class, new ByteArrayLoader.StreamFactory())
           /* Transcoders */
           .register(Bitmap.class, BitmapDrawable.class,
               new BitmapDrawableTranscoder(resources, bitmapPool))
           .register(Bitmap.class, byte[].class, new BitmapBytesTranscoder())
           .register(GifDrawable.class, byte[].class, new GifDrawableBytesTranscoder());

2. onDataReady() 콜백
3. 요청했던 size대로 다시 encoding하여 캐쉬에 넣는다.

 - 이것을 도식화해보면 아래와 같다. (DataFetcher를 상속받은 클래스가 너무 많아 몇개는 생략하였다)