INTERNAL: make piped insert operations process synchronously

[oliviarla]

🔗 Related Issue

• https://github.com/jam2in/arcus-works/issues/540

⌨️ What I did

• lop/sop/mop/bop piped insert에 한해 적용하는 PR입니다.

동작 과정

• 500개 아이템 단위로 Operation 객체를 나누어 비동기로 일제히 요청을 보내던 것을 동기 방식으로 하나씩 보내어 Arcus 서버에 과도한 부하가 들어가는 것을 방지하고, 실패 시 다음 Operation을 수행하지 않도록 합니다.

• 이전 operation이 성공해야만 다음 operation이 writeQueue와 Future의 operation list에 추가됩니다.

• 만약 CLIENT_ERROR, SERVER_ERROR 실패가 발생하면 그 즉시 latch.countdown을 호출해 남은 operation을 수행하지 않습니다.

• OVERFLOWED, OUT_OF_RANGE 같은 실패가 발생하면 이후 operation의 첫 command가 NOT_EXECUTED 상태임을 future의 failedResult에 추가합니다.

• NOT_EXECUTED 이후의 모든 연산은 실행되지 않은 것입니다. (CANCELED가 아닌 NOT_EXECUTED를 사용하는 이유는 진짜 cancel 상황과의 혼용을 막고자 하기 위함입니다.)

• 본 PR 이전에 pipe 관련 클래스 구조 변경이 있었고, 이에 맞추어 SingleKeyPipedOperationImpl 클래스를 추가합니다.

  • PipedOperationImpl 자체를 수정할 경우 bulk insert까지 영향이 가기 때문에, single key 전용 operation 클래스를 추가한 후 handleLine 메서드를 수정했습니다.
  • 이 클래스를 CollectionPipedInsertOperationImpl 가 상속받도록 합니다. (추후 CollectionPipedUpdateOperationImpl 등 클래스들도 상속받을 예정입니다.)
  • PipedOperationImpl 클래스는 파이프를 사용하는 모든 Operation에 대한 로직을 관리하고, single key나 multi key에 대한 처리를 이 클래스와 같은 별도 클래스에 두어 분리된 형태로 관리할 수 있습니다.

[oliviarla]

@jhpark816 리뷰 반영했습니다.

[jhpark816]

@uhm0311 리뷰 바랍니다.

[uhm0311]

@oliviarla

컨플릭트 확인해주세요.

[oliviarla]

@uhm0311 @jhpark816 첫 코멘트에 새롭게 구현된 사항을 다시 정리해두었습니다. 리뷰 부탁드립니다.

[oliviarla]

cancel 관련 설명

• 기존에는 future를 통한 cancel 요청이 "operation을 complete 처리중이고 nextOp가 남아있었다면" nextOp를 cancel하는 대신 future를 통한 cancel을 실패했다고 처리하였습니다.
• cancel 요청이 들어오면 현재 진행중인 op를 cancel시키거나 현재 진행중인 op를 complete 처리중이라면 nextOp를 cancel 시키도록 처리하는 것이 적절합니다.
• 따라서 위와 같은 방식으로 cancel 요청을 처리하고자 합니다.
  • 참고로 nextOp의 cancel 처리를 future에서 진행해야 cancel 완료 여부를 쉽게 바로 확인할 수 있으므로, future에 완료되었거나 수행중인 Op 뿐만 아니라 수행 대기중인 모든 op까지 담도록 변경하겠습니다.

처리 흐름

• 다음은 future.cancel()이 호출되는 시점에 따른 케이스를 대략 정리해보았습니다. future.cancel() 로직과 메서드 로직은 동시에 실행될 수 있어 호출 시점과 로직 수행 시점이 다를 수 있지만 일단은 케이스를 나눠 생각해보겠습니다.

  • 호출 시점은 코드에 /* <case N> */ 로 적어두었습니다.

  • future.cancel() 은 대략 아래와 같이 동작할 것입니다.

    public boolean cancel(boolean ign) {
       this.startCancel(); // canceling flag를 true로 설정
       boolean succeed;
       if (ops.get(currentOpIdx).cancel(ign)) {
         succeed = true;
       } else {
           this.lock();
           if (currentOpIdx < ops.size() - 1) {
             succeed = ops.get(++currentOpIdx).cancel(ign);
           } else {
             succeed = false;
           }
           this.unlock();
       }
       this.endCancel(); // canceling flag를 false로 설정
       return succeed;
    }
    

future.cancel() 호출이 transitionState(OperationState.COMPLETE) 호출과 겹칠 때

protected final void transitionState(OperationState newState) {
  state = newState;
  if (state != OperationState.WRITE_QUEUED &&
      state != OperationState.WRITING) {
    cmd = null;
  }
  /* <case 1> */
  if (state == OperationState.COMPLETE &&
          callbacked.compareAndSet(false, true)) {
    /* <case 2> */
    callback.complete();
    /* <case 3> */
  }
}

• <case 1>
  • state는 COMPLETE이지만 cancel() 메서드 수행 시 callbacked를 set 하게 되어, cancel 요청은 성공한다.
• <case 2>
  • 해당 op는 완료된 상태이므로 cancel() 메서드 수행에 실패한다.
  • future.cancel() 에서 남은 op가 있는 경우 cancel 시켜 callback.complete() 에서 더 이상 op가 input queue에 등록되지 않도록 한다.
  • future.cancel() 에서 남은 op가 없는 경우 cancel 요청을 실패 처리 한다.
• <case 3>
  • 이미 callback을 통해 nextOp가 등록된 시점이므로, 수행중인 operation에 대한 cancel 요청이 성공할 것이다.

future.cancel() 호출이 callback.complete()와 겹칠 때

public void complete() {
  if (rv.getOperationStatus().isSuccess()) {
    Operation op = rv.getNextOp(); // next op가 있다면 op 객체를, 없다면 Null 반환
    if (op != null) {
      Operation nextOp = ops.get(nextOpIdx);
      /* <case 4> */
      rv.lock();
      if (!nextOp.isCancelled() && !future.isCanceling()) {
        addOp(key, nextOp);
        rv.setCurrentOpIdx(nextOpIdx);
      }
      rv.unlock();
    } else {
      /* <case 5> */
      latch.countDown();
    }
  } else {
    /* <case 6> */
    // ...
    if (nextIndex > 0) {
      rv.addEachResult(nextIndex,
              new CollectionOperationStatus(false, "NOT_EXECUTED", CollectionResponse.NOT_EXECUTED));
    }
    latch.countDown();
  }
}

• <case 4>
  • 현재 op가 complete되었고 nextOp를 input queue에 넣기 직전 future.cancel() 요청이 들어온 상황이다.
  • future.cancel 수행 시작 시점에 canceling flag를 true로 두고, nextOp cancel 작업 도중에는 nextOp를 input queue에 추가하지 않도록 lock을 사용한다.
• <case 5>
  • 현재 op가 complete되었고 nextOp가 없으므로 cancel 요청은 실패한다.
• <case 6>
  • 현재 op가 실패했고 nextOp를 실행하지 않는다는 flag를 세워야 하므로 cancel 요청은 실패한다.

[uhm0311]

@oliviarla

코멘트만 봐서는 잘 이해가 되지 않습니다. 오프라인으로 더 자세히 설명해주실 수 있나요?

[oliviarla]

@uhm0311 @jhpark816 코멘트 내용을 좀더 자세히 수정하였습니다. 검토 부탁드립니다.

[uhm0311]

기존 PR의 어떤 부분이 어떻게 문제가 되었는지부터가 이해가 안되는 부분이라서, 거기서부터 설명이 필요합니다. 지금 코멘트는 TO-BE에 관한 설명만 있습니다. AS-IS의 문제가 무엇이었는지, 어떠한 이유로 문제였는지부터 천천히 설명해주세요.

기존 cancel 처리 과정에서 operation을 complete 처리중일 때 future로부터 cancel을 요청했고 nextOp가 남아있었다면 이를 cancel하는 대신 실패 응답을 보내고 있었습니다.

실패 응답이 무엇인가요?

• cancel() 메소드가 실패했다는 응답?
• 캐시 연산이 실패했다는 응답?
• 다른 기타 응답?

[oliviarla]

@uhm0311 원본 코멘트에 더 자세히 작성했습니다. cancel()메서드가 실패했다는 응답을 의미했습니다.

[oliviarla]

@uhm0311 @jhpark816 이전 코멘트가 너무 복잡한 방향으로 진행되어 @jhpark816 님의 피드백을 받고 아래와 같이 구상을 변경하였습니다.

future.cancel()

• op.cancel이 실패하는 경우는 이미 op가 complete 처리되고 있는 경우 뿐이다. 따라서 complete 처리된 op는 스킵하고 다음 op를 cancel 시도한다.
• 현재 수행중인 op부터 순회하며 cancel이 성공하였다면 즉시 성공 응답을 반환하도록 한다.
• cancel된 op가 하나라도 있다면 future는 cancel된 상태이다.

public boolean cancel(boolean ign) {
  for (int i = currentOpIdx; i < ops.size(); i++) {
    if (ops.get(i).cancel("by application.")) {
      return true;
    }
  }
  return false;
}

public boolean isCancelled() {
  for (int i=0;i<ops.size();i++) {
    if (ops.get(i).isCancelled()) {
      return true;
    }
  }
  return false;
}

callback.complete()

• nextOp를 cancel 시키는 동시에 addOp하는 경우 cancel 상태이면서 input queue에 추가될 수 있으나, write queue에서 제외되므로 문제가 발생하지 않는다.

public void complete() {
  if (rv.getOperationStatus().isSuccess()) {
    Operation nextOp = rv.getNextOp(); // next op가 있다면 op 객체를, 없다면 Null 반환
    if (nextOp != null && !nextOp.isCancelled()) {
      addOp(key, nextOp);
      rv.setCurrentOpIdx(nextOpIdx);
    } else {
      latch.countDown();
    }
  } else {
    // ...
    if (nextIndex > 0) {
      rv.addEachResult(nextIndex, new CollectionOperationStatus(false, "NOT_EXECUTED", CollectionResponse.NOT_EXECUTED));
    }
    latch.countDown();
  }
}

[oliviarla]

@uhm0311 @jhpark816 이제부터 다시 리뷰해주시면 됩니다.

[oliviarla]

리뷰 반영했습니다.

[oliviarla]

@uhm0311 @jhpark816 현재 Future에 currentOpIdx, 모든 Operation을 담은 List 를 두다보니 해당 값을 callback에서도 사용하고 future에서도 사용하여, 동시성 문제가 발생할 여지가 높아져있습니다. 또한 Future는 값을 기다리는 역할만을 갖는 것이 단순하고 추후 Future를 통합할 때도 유용할 것이라는 생각이 들었습니다.

따라서 Future에 해당 값들을 두지 않고, 기존에 수행했던대로 Future에는 현재 실행중이거나 실행 완료된 Operation만을 담는 List를 갖도록 변경하는 PR을 다시 올리려고 합니다.

기존에서 현 상태로 구조가 바뀌게 된 원인이 future.cancel() 호출 시 nextOp까지 cancel() 해보지 않고 실패 처리하여, nextOp에 쉽게 접근해 cancel() 호출을 하기 위함이었습니다.

하지만 이 문제는 Future에서 직접 nextOp에 접근하는 방식 대신, 하나의 Operation에 next Operation 변수를 담아 Operation에 cancel 요청이 들어왔을 때 next Operation cancel()을 내부적으로 호출할 수 있습니다. 구현은 완료된 상태이고, 자체 리뷰 거쳐 내일 다시 PR 올리겠습니다.