INTERNAL: make lop piped operations process synchronously

[oliviarla]

🔗 Related Issue

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

⌨️ What I did

• 500개 아이템 단위로 Operation 객체를 나누어 비동기로 요청을 보내던 것을 동기 방식으로 보내어 Arcus 서버에 과도한 부하가 들어가는 것을 방지하고, 실패 시 다음 Operation을 수행하지 않도록 합니다.
  • 만약 CLIENT_ERROR, SERVER_ERROR 실패가 발생하면 그 즉시 latch.countdown을 호출해 더이상 로직을 수행하지 않습니다.
  • OVERFLOWED, OUT_OF_RANGE 같은 실패가 발생하면 이후의 item에 대해 CANCELLED 오류를 반환하도록 합니다.
• 기존에 asyncLopPipedInsert에서 입력받은 valueList 순서대로 저장하지 않는 문제가 있었고 이를 순서대로 저장하도록 변경했습니다.
• 시범적으로 Lop Piped Insert에만 적용합니다.
• future를 통한 비동기 수행이 되기 때문에 메서드명은 그대로 두고자 합니다.

[jhpark816]

일부 리뷰

[jhpark816]

rv.getOperationStatus().isSuccess() 조건으로 next 연산 수행 여부를 판단하기 어렵습니다.
이 부분에 대해 offline 논의해야 할 것 같습니다.

[jhpark816]

관련 사항을 좀 더 자세하게 적으면 다음과 같습니다.

• PIPE_ERROR 응답이 온 경우
  • 현재의 piped 연산에서 PIPE_ERROR로 인해 아예 수행되지 item 연산이 있을 수 있습니다.
  • 이 경우, 현재 failed Result에 결과를 담지 않고 있으므로, 이를 추가해야 합니다.
• PIPE_END 응답이 온 경우
  • 개별 연산에 대해 NOT_FOUND, ELEMENT_EXISTS, OVERFLOWED, OUT_OF_RANGE, TYPE_MISMATCH, BKEY_MISMATCH 오류가 발생할 수 있습니다.
  • B+Tree collection에 대해 BKEY_MISMATCH 오류가 발생한 경우, 주어진 element만 insert 실패하고 나머진 성공할 수 있습니다. 이 경우, next 연산을 수행하는 것이 맞지 않는 지 ? ELEMENT_EXISTS 오류도 비슷해 보입니다.
  • 그 외의 오류인 경우는 캐시 서버에서 수행을 중지하고, PIPE_ERROR 리턴해야 하지 않나 생각합니다.
  • 결국, 캐시 서버 동작 부분도 함께 고려되어야 하지 않나 생각합니다.

[oliviarla]

END 응답이 왔고 btree일 때 BKEY_MISMATCH, map, set일 때 ELEMENT_EXISTS만 발생했다면 다음 Operation 보내는 것에는 동의합니다.

PIPE_ERROR가 CLIENT_ERROR, SERVER_ERROR 로 인해 발생했다면 예외가 발생해 failed result를 조회할 수 없습니다. PIPE_ERROR가 command/memory overflow 로 인해 발생했다면 에러를 무시하고 모든 응답을 기준으로 전체 성공/실패 여부만 판별합니다. 따라서 "PIPE_ERROR" 라는 응답이 왔는지 future에서는 확인할 수 없으며, failed result에 결과를 넣을 필요도 없습니다.

arcus-java-client/src/main/java/net/spy/memcached/protocol/ascii/CollectionPipedInsertOperationImpl.java

Lines 139 to 148 in f60ed48

	if (line.startsWith("END") || line.startsWith("PIPE_ERROR ")) {
	/* ENABLE_MIGRATION if */
	if (needRedirect()) {
	transitionState(OperationState.REDIRECT);
	return;
	}
	/* ENABLE_MIGRATION end */
	cb.receivedStatus((successAll) ? END : FAILED_END);
	transitionState(OperationState.COMPLETE);
	} else if (line.startsWith("RESPONSE ")) {

[jhpark816]

@oliviarla

기존 pipe 처리에도 일부 문제가 있는 것 같습니다.

• CLIENT_ERROR, SERVER_ERROR 등의 오류가 있는 경우, java client는 PIPE_ERROR 응답까지 읽지 않는다.
  그러면, 다음 연산에서 응답을 읽을 시에 PIPE_ERROR 응답을 보게 되는 문제가 있습니다.
  대략 확인한 사항이라, 이 부분에 관한 코드는 직접 확인해 보기 바랍니다.
• 아래 2가지 PIPE_ERROR에 대해 "END" 응답을 받은 경우와 동일하게 처리합니다.
  즉, 처리하지 못한 연산이 존재하더라도 처리가 완료된 것으로 판단하고 있습니다.
  • PIPE_ERROR command overflow
  • PIPE_ERROR memory overflow

이러한 pipe 처리 오류는 본 PR과는 별도로 처리해야 하죠?
검토 바랍니다.

[oliviarla]

CLIENT_ERROR, SERVER_ERROR 등의 오류가 있는 경우, 바로 OperationException이 throw 됩니다.
이 예외를 MemcachedConnection에서 catch하여 연결을 끊기 때문에 다음 연산이 응답을 읽을 수 없을 것으로 보입니다.

arcus-java-client/src/main/java/net/spy/memcached/MemcachedConnection.java

Lines 948 to 952 in 45720de

	} catch (OperationException e) {
	qa.setupForAuth("operation exception"); // noop if !shouldAuth
	getLogger().warn("Reconnection due to exception " +
	"handling a memcached exception on %s.", qa, e);
	lostConnection(qa, ReconnDelay.IMMEDIATE, "operation exception");

[jhpark816]

PIPE_ERROR command/memory overflow 오류 시, 처리되지 못한 연산에 대해
본 PR 처럼 CANCELED CollectionOperationStatus 설정하는 것이 좋겠습니다.

[jhpark816]

PIPE_ERROR command/memory overflow 오류 시, 처리되지 못한 연산에 대해
본 PR 처럼 CANCELED CollectionOperationStatus 설정하는 것이 좋겠습니다.

위 사항을 구현하기 위해
아래 로직에서 cb.receivedStatus() 호출 시의 인자 값이 수정되어야 할 것 같습니다.

         if (line.startsWith("END") || line.startsWith("PIPE_ERROR ")) { 
	   /* ENABLE_MIGRATION if */ 
	   if (needRedirect()) { 
	     transitionState(OperationState.REDIRECT); 
	     return; 
	   } 
	   /* ENABLE_MIGRATION end */ 
	   cb.receivedStatus((successAll) ? END : FAILED_END); 
	   transitionState(OperationState.COMPLETE); 
	 } else if (line.startsWith("RESPONSE ")) { 

아래와 같이 호출해야 할 것 같습니다.

	   cb.receivedStatus((index == count && successAll) ? END : FAILED_END); 

또한, 아래 경우에 CENCEL 상태를 추가하는 로직도 수정되어야 합니다.

        if (!rv.getOperationStatus().isSuccess()) {
          rv.addEachResult(index + (idx * CollectionPipedInsert.MAX_PIPED_ITEM_COUNT),
                  new CollectionOperationStatus(false, "CANCELED", CollectionResponse.CANCELED));
          latch.countDown();
        } 

위와 같이 변경되면,
complete() 로직에서의 검사 순서도 변경되어야 합니다.
아래 조건이 가장 먼저 검사되어야 할 것입니다.

        if (!rv.getOperationStatus().isSuccess()) 

[jhpark816]

리뷰 완료

[oliviarla]

@jhpark816 리뷰 반영했습니다.

[jhpark816]

@uhm0311 리뷰 바랍니다.

[uhm0311]

@oliviarla

컨플릭트 확인해주세요.

[jhpark816]

리뷰 완료

[jhpark816]

complete() 로직을 완벽하게 만들려면,
PIPE_ERROR이더라도 모든 응답을 읽어내는 기능이 먼저 구현되어야 할 것 같습니다.
이를 이슈로 올리고 작업을 시작하면 좋겠습니다.

[jhpark816]

PIPE_ERROR command/memory overflow 오류 시, 처리되지 못한 연산에 대해
본 PR 처럼 CANCELED CollectionOperationStatus 설정하는 것이 좋겠습니다.

위 사항을 구현하기 위해
아래 로직에서 cb.receivedStatus() 호출 시의 인자 값이 수정되어야 할 것 같습니다.

         if (line.startsWith("END") || line.startsWith("PIPE_ERROR ")) { 
	   /* ENABLE_MIGRATION if */ 
	   if (needRedirect()) { 
	     transitionState(OperationState.REDIRECT); 
	     return; 
	   } 
	   /* ENABLE_MIGRATION end */ 
	   cb.receivedStatus((successAll) ? END : FAILED_END); 
	   transitionState(OperationState.COMPLETE); 
	 } else if (line.startsWith("RESPONSE ")) { 

아래와 같이 호출해야 할 것 같습니다.

	   cb.receivedStatus((index == count && successAll) ? END : FAILED_END); 

또한, 아래 경우에 CENCEL 상태를 추가하는 로직도 수정되어야 합니다.

        if (!rv.getOperationStatus().isSuccess()) {
          rv.addEachResult(index + (idx * CollectionPipedInsert.MAX_PIPED_ITEM_COUNT),
                  new CollectionOperationStatus(false, "CANCELED", CollectionResponse.CANCELED));
          latch.countDown();
        } 

위와 같이 변경되면,
complete() 로직에서의 검사 순서도 변경되어야 합니다.
아래 조건이 가장 먼저 검사되어야 할 것입니다.

        if (!rv.getOperationStatus().isSuccess())