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에 대한 처리를 이 클래스와 같은 별도 클래스에 두어 분리된 형태로 관리할 수 있습니다.

[jhpark816]

일부 리뷰

[jhpark816]

리뷰 완료

[oliviarla]

@jhpark816 리뷰 반영했습니다.

[jhpark816]

@uhm0311 리뷰 바랍니다.

[uhm0311]

하나 더 질문입니다.

[uhm0311]

세 번째 질문 남깁니다.

[jhpark816]

일부 리뷰

[jhpark816]

일부 리뷰

[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
이제부터 다시 리뷰해주시면 됩니다.

[uhm0311]

nextCommandIdx 값이 0인 경우에는 왜 처리하지 않아도 되는지를 설명하는 주석을 추가해주세요.

[uhm0311]

currentCommandIdx 값이 -1인 경우에 대해서는 따로 처리를 하지 않아도 되나요?

[jhpark816]

리뷰 완료

[jhpark816]

complete()는 아래 정도의 로직이 나을 것 같습니다.

• getNextOperation()는 cancel 되지 않은 next op를 리턴하는 함수를 두도록 합시다.
• currentCommandIdx => currItemIdx (op 내의 n개 items 중에서 하나를 가리키는 index)
• this.gotStatus() 호출하면 코드가 간단해 짐.

        if (rv.getOperationStatus().isSuccess()) {
          Operation nextOp = rv.getNextOperation();
          if (nextOp != null) {
            addOp(key, nextOp);
            return;
          }
        } else {
          if ((currItemIdx + 1) < itemCount || (opIdx + 1) < insertList.size()) {
            this.gotStatus(currItemIdx+1, 
                           new CollectionOperationStatus(false, "NOT_EXECUTED", 
                               CollectionResponse.NOT_EXECUTED));
          }
        }
        latch.countDown();

[jhpark816]

NOT_EXECUTED => STOPPED 로 할까요?

[jhpark816]

isCancelled()에서도 아래 for 문을 사용하는 것이 나은 지를 검토 바랍니다.

for (int i = currentOpIdx; i < ops.size(); i++)