🏛️

[FP & JS ES6+] 5. 비동기: 동시성 프로그래밍 (2) - 지연평가 + Promise, 병렬적 평가

1. 지연평가 + Promise

1.1. L.map, map, take

// go1
const go1 = (a, f) => a instanceof Promise ? a.then(f) : f(a);

// L.map
L.map = curry(function* (f, iter) {
  for (const a of iter) {
    yield go1(a, f);
  }
});

// take 
export const take = curry((l, iter) => {
  let res = [];
  iter = iter[Symbol.iterator]();
  return function recur() {
    let cur;
    while (!(cur = iter.next()).done) {
      const a = cur.value;
      if (a instanceof Promise) {
        return a.then(a => {
          res.push(a);
          if (res.length === l) return res;
          return recur();
        });
      }
      res.push(a);
      if (res.length === l) return res;
    }

    return res;
  } ()
});

// take all
export const takeAll = take(Infinity);

// map
export const map = curry(pipe(L.map, takeAll));


JavaScript
복사

•

적용 코드

import {log, go, L, take, map, takeAll} from '../0_common/fx.js';

go(
  [Promise.resolve(1), Promise.resolve(2), Promise.resolve(3)],
  map(a => Promise.resolve(a + 10)),
  take(2),
  log);

go(
  [1, 2, 3],
  map(a => Promise.resolve(a + 10)),
  takeAll,
  log);
JavaScript
복사

1.2. Kleisli Composition - L.filter, filter, nop, take

•

L.filter & take에 promise 적용

•

비동기 동시성과 지연평가가 모두 가능한 코드 구현

const nop = Symbol('nop');

// go
const go1 = (a, f) => a instanceof Promise ? a.then(f) : f(a);

// L.map
L.map = curry(function* (f, iter) {
  for (const a of iter) {
    yield go1(a, f);
  }
});

// L.filter
L.filter = curry(function* (f, iter) {
  for (const a of iter) {
    const b = go1(a, f);
    if (b instanceof Promise) yield b.then(b => b ? a : Promise.reject(nop));
    else if (b) yield a;
  }
});


// take 
export const take = curry((l, iter) => {
  let res = [];
  iter = iter[Symbol.iterator]();
  return function recur() {
    let cur;
    while (!(cur = iter.next()).done) {
      const a = cur.value;
      if (a instanceof Promise) {
        return a.then(a => {
          res.push(a);
          if (res.length === l) return res;
          return recur();
        }).catch(e => e === nop ? recur() : Promise.reject(e));
      }
      res.push(a);
      if (res.length === l) return res;
    }

    return res;
  }()
});
JavaScript
복사

•

적용 코드

import {go, L, log, take} from '../0_common/fx.js';

// Kleisli Composition - L.filter, filter, nop, take

go([1, 2, 3, 4, 5, 6],
  L.map(a => Promise.resolve(a * a)),
  L.filter(a =>  Promise.resolve(a % 2)),
  L.map(a => Promise.resolve(a * a)),
  take(2),
  log
);

go([1, 2, 3, 4, 5, 6],
  L.map(a => a * a),
  L.filter(a =>  a % 2),
  L.map(a => a * a),
  take(2),
  log
);
JavaScript
복사

1.3. reduce - nop 지원

•

reduce에 promise 적용

◦

before

export const reduce = curry((f, acc, iter) => {
  if (!iter) {
    iter = acc[Symbol.iterator]();
    acc = iter.next().value
  }
  return go1(acc, function recur(acc) {
    let cur;
    while (!(cur = iter.next()).done) {
      acc = f(acc, a);
      if (acc instanceof Promise) return acc.then(recur);
    }
    return acc;
  });
});
JavaScript
복사

◦

after

const reduceF = (acc, a, f) =>
  a instanceof Promise ?
    a.then(a => f(acc, a), e => e === nop ? acc : Promise.reject(e)) :
    f(acc, a);

export const reduce = curry((f, acc, iter) => {
  if (!iter) {
    iter = acc[Symbol.iterator]();
    acc = iter.next().value
  }
  return go1(acc, function recur(acc) {
    let cur;
    while (!(cur = iter.next()).done) {
      acc = reduceF(acc, cur.value, f);
      if (acc instanceof Promise) return acc.then(recur);
    }
    return acc;
  });
});
JavaScript
복사

◦

iter가 없는 경우 함수 표현식으로 리팩토링

const reduceF = (acc, a, f) =>
  a instanceof Promise ?
    a.then(a => f(acc, a), e => e === nop ? acc : Promise.reject(e)) :
    f(acc, a);

const head = iter => go1(take(1, iter), ([h]) => h);

export const reduce = curry((f, acc, iter) => {
  if (!iter) {
    return reduce(f, head(iter = acc[Symbol.iterator]()), iter);
  }
  return go1(acc, function recur(acc) {
    let cur;
    while (!(cur = iter.next()).done) {
      acc = reduceF(acc, cur.value, f);
      if (acc instanceof Promise) return acc.then(recur);
    }
    return acc;
  });
});
JavaScript
복사

•

적용 코드

import {go, L, log, reduce, takeAll} from '../0_common/fx.js';

// reduce에서 nop 지원

const add = (a, b) => a + b;

go([1, 2, 3, 4],
  L.map(a => Promise.resolve(a * a)),
  L.filter(a => Promise.resolve(a % 2)),
  reduce(add),
  log // 25
);
JavaScript
복사

1.4. 지연평가 + Promise의 효율성

•

takeAll로 대략 16초가 걸릴 로직도 take(2)만 할경우 4초 내로 로직 수행 가능

import {go, L, log, take, takeAll, map, filter} from '../0_common/fx.js';


go([1, 2, 3, 4, 5, 6, 7, 8],
  L.map(a => {
    log(a);
    return new Promise(resolve => setTimeout(() => resolve(a * a), 1000));
  }),
  L.filter(a => {
    log(a);
    return new Promise(resolve => setTimeout(() => resolve(a % 2), 1000));
  }),
  takeAll,
  log
);

go([1, 2, 3, 4, 5, 6, 7, 8],
  L.map(a => {
    log(a);
    return new Promise(resolve => setTimeout(() => resolve(a * a), 1000));
  }),
  L.filter(a => {
    log(a);
    return new Promise(resolve => setTimeout(() => resolve(a % 2), 1000));
  }),
  take(2),
  log
);
JavaScript
복사

2. 지연된 함수열을 병렬적으로 평가하기

2.1. C.reduce

•

비동기성을 고려하지 않고, 모든 배열을 평가해 reduce로 넘김

•

적용 코드

◦

L.map과 L.filter를 사용하므로 go함수 내부 reduce에서 함수별로 하나씩 동기적으로 delay1000 함수 실행

◦

C.reduce를 활용하면 배열을 펼쳐 한 번에 평가 후 다음 함수로 전달

import {curry, go, L, log, reduce} from '../0_common/fx.js';


// 지연된 함수열을 병렬적으로 평가하기 - C.reduce, C.take

export const C = {};
C.reduce = curry((f, acc, iter) => iter ?
  reduce(f, acc, [...iter]) :
  reduce(f, [...acc]));


const add = (a, b) => a + b;

const delay1000 = a => new Promise(resolve =>
  setTimeout(() => resolve(a), 1000));


go([1, 2, 3, 4, 5],
  L.map(a => delay1000(a * a)),
  L.filter(a => a % 2),
  // reduce(add),
  C.reduce(add),
  log
);
JavaScript
복사

2.2. Promise.reject의 평가

•

병렬적으로 함수를 실행하다보면 비동기적으로 promise reject가 발생해 작업을 처리해줘야하는 경우가 발생한다.

•

처리를 따로 안하더라도 코드는 동작하나, console에 에러가 찍힘.

•

이를 방지하기 위해 promise reject를 catch하여 처리 진행

•

단, 이때 catch된 iterator는 또다시 catch하는것은 불가능하다.

•

적용 코드

import {curry, go, L, log, reduce} from '../0_common/fx.js';


// Promise reject 평가

export const C = {};

function noop() {}
const catchNoop = arr => (arr.forEach(a => a instanceof Promise ? a.catch(noop) : a), arr);

C.reduce = curry((f, acc, iter) => {
  const iter2 = catchNoop(iter ? [...iter] : [...acc]);
  return iter ?
    reduce(f, acc, iter2) :
  reduce(f, iter2);
});

const add = (a, b) => a + b;

const delay1000 = a => new Promise(resolve =>
  setTimeout(() => resolve(a), 1000));


go([1, 2, 3, 4, 5,6,7,8,9],
  L.map(a => delay1000(a * a)),
  L.filter(a => delay1000(a % 2)),
  L.map(a => delay1000(a * a)),
  C.reduce(add),
  log
);
JavaScript
복사

2.3. C.take

•

마찬가지로 배열을 병렬 처리 하기 위해 spread 한뒤 즉시 평가 후 catchNoop에 전달

•

적용 코드

import {curry, go, L, log, reduce, take} from '../0_common/fx.js';


// C.take

export const C = {};

function noop() {
}

const catchNoop = arr => (arr.forEach(a => a instanceof Promise ? a.catch(noop) : a), arr);

C.reduce = curry((f, acc, iter) => {
  const iter2 = catchNoop(iter ? [...iter] : [...acc]);
  return iter ?
    reduce(f, acc, iter2) :
    reduce(f, iter2);
});

C.take = curry((l, iter) => {
  return take(l, catchNoop([...iter]));
});

const add = (a, b) => a + b;

const delay1000 = a => new Promise(resolve =>
  setTimeout(() => resolve(a), 1000));


go([1, 2, 3, 4, 5, 6, 7, 8, 9],
  L.map(a => delay1000(a * a)),
  L.filter(a => delay1000(a % 2)),
  L.map(a => delay1000(a * a)),
  C.take(2),
  C.reduce(add),
  log
);
JavaScript
복사

2.4. C.map, C.filter

•

병렬적으로 즉시 평가 진행

C.map = curry(pipe(L.map, C.takeAll));
C.filter = curry(pipe(L.filter, C.takeAll));

const delay1000 = a => new Promise(resolve =>
  setTimeout(() => resolve(a), 1000));

C.map(a => delay1000(a * a), [1, 2, 3, 4]).then(log);
C.filter(a => delay1000(a % a), [1, 2, 3, 4]).then(log);
JavaScript
복사

3. 즉시, 지연, Promise, 병렬적 조합하기

3.1. 즉시 평가

•

모든 평가를 즉시, 엄격하게 실행

•

각 함수들이 종(왼쪽 → 오른쪽)으로 실행되고 완료된 이후 다음 함수로 이동함

•

소요시간 약 9초

const delay500 = (a, name) => new Promise(resolve => {
  console.log(`${name}: ${a}`);
  setTimeout(() => resolve(a), 500);
});

console.time('');
go([1, 2, 3, 4, 5, 6, 7, 8, 9],
  map(a => delay500(a * a, 'map 1')),
  filter(a => delay500(a % 2, 'filter 2')),
  map(a => delay500(a + 1, 'map 3')),
  take(2),
  log,
  _ => console.timeEnd(''));
JavaScript
복사

•

출력된 결과

map 1: 1
map 1: 4
map 1: 9
map 1: 16
map 1: 25
filter 2: 1
filter 2: 0
filter 2: 1
filter 2: 0
filter 2: 1
map 3: 2
map 3: 5
map 3: 10
map 3: 17
map 3: 26
[ 2, 5 ]
JavaScript
복사

3.2. 지연 평가

•

모든 평가를 지연(Lazy)하게 진행

•

각 함수들의 평가를 횡(위 → 아래)로 실행

•

평가를 최소화하는 방향으로 동작 

•

소요시간 약 4초

console.time('');
go([1, 2, 3, 4, 5],
  L.map(a => delay500(a * a, 'map 1')),
  L.filter(a => delay500(a % 2, 'filter 2')),
  L.map(a => delay500(a + 1, 'map 3')),
  take(2),
  log,
  _ => console.timeEnd(''));
JavaScript
복사

•

출력된 결과

map 1: 1
filter 2: 1
map 3: 2

map 1: 4
filter 2: 0

map 1: 9
filter 2: 1
map 3: 10
JavaScript
복사

3.3. 병렬적으로 평가

•

특정 로직을 병렬적으로 즉시 실행하게끔 적용 가능

•

아래 예시에서는 첫번째 map 함수를 병렬적으로 평가 진행

•

소요시간 약 3초

console.time('');
go([1, 2, 3, 4, 5],
  C.map(a => delay500(a * a, 'map 1')),
  L.filter(a => delay500(a % 2, 'filter 2')),
  L.map(a => delay500(a + 1, 'map 3')),
  take(2),
  log,
  _ => console.timeEnd(''));
JavaScript
복사

•

출력된 결과

map 1: 1
map 1: 4
map 1: 9
map 1: 16
map 1: 25

filter 2: 1
map 3: 2

filter 2: 0

filter 2: 1
map 3: 10
JavaScript
복사

3.4. 다양한 조합

console.time('');
go([1, 2, 3, 4, 5],
  L.map(a => delay500(a * a, 'map 1')),
  C.filter(a => delay500(a % 2, 'filter 2')),
  L.map(a => delay500(a + 1, 'map 3')),
  take(2),
  log,
  _ => console.timeEnd(''));


map 1: 1
map 1: 4
map 1: 9
map 1: 16
map 1: 25
filter 2: 1
filter 2: 0
filter 2: 1
filter 2: 0
filter 2: 1
map 3: 2
map 3: 10
JavaScript
복사

console.time('');
go([1, 2, 3, 4, 5],
  L.map(a => delay500(a * a, 'map 1')),
  L.filter(a => delay500(a % 2, 'filter 2')),
  L.map(a => delay500(a + 1, 'map 3')),
  C.take(2),
  log,
  _ => console.timeEnd(''));


map 1: 1
map 1: 4
map 1: 9
map 1: 16
map 1: 25
filter 2: 1
filter 2: 0
filter 2: 1
filter 2: 0
filter 2: 1
map 3: 2
map 3: 10
map 3: 26
JavaScript
복사