uk csharp 6.task

This is an old version of this page. You can view the most recent version or browse the history.

Threading

Thread는 Process 내부에서 생성되는 명령어를 실행하기 위한 스케줄링의 단위이다. Process를 생성할 때 기본적으로 한 개의 Main Thread를 생성한다. 한 개의 Thread에서 복잡한 작업을 하고 있다면, 그 작업이 끝날 때까지 다른 작업은 할 수 없게 된다. 이 경우 별도의 Thread를 생성하여 작업을 분담시킴으로써 문제를 해소할 수 있다.

System.Threading

Thread

Thread를 생성하고 실행하는 데 필요한 명령어를 가진 클래스다.

Foreground Thread

Foreground Thread는 Main Thread의 종료와 상관없이 모든 Thread가 종료되어야 Process가 종료된다.

static void Main()
{
	Console.WriteLine("Main Thread Start");
	
	Thread thread = new Thread(threadFunc);
	thread.Start();
	
	Console.WriteLine("Main Thread End");
}

static void threadFunc()
{
	Console.WriteLine("Foreground Thread Start");
	
	Thread.Sleep(3000); // 3초 동안 실행 중지
	
	Console.WriteLine("Foreground Thread End");
}

실행 결과 : 
Main Thread Start
Foreground Thread Start
Main Thread End
Foreground Thread End

Background Thread

Background Thread는 Thread의 종료 여부와 상관없이 Main Thread가 종료되면 Process도 종료된다. 단, Join()을 사용하면 해당 Thread가 종료할 때까지 현재 Thread는 대기한다.

static void Main()
{
	Console.WriteLine("Main Thread Start");
	
	Thread thread = new Thread(threadFunc);
	thread.IsBackground = true;
	thread.Start();
	
	thread.Join(); // thread가 종료할 때까지 대기
	
	Console.WriteLine("Main Thread End");
}

static void threadFunc()
{
	Console.WriteLine("Background Thread Start");
	
	Thread.Sleep(3000); // 3초 동안 실행 중지
	
	Console.WriteLine("Background Thread End");
}

실행 결과 : 
Main Thread Start
Background Thread Start
Background Thread End
Main Thread End

Monitor

Monitor는 임의의 객체를 잠금으로써 특정 코드의 접근을 막는 기능을 가지고 있다. 여러 Thread가 하나의 공유자원을 동시에 사용하면 문제가 발생하는데 Monitor를 통해 하나의 Thread에서만 사용할 수 있도록 하여 Thread를 동기화할 수 있다.

Count를 1씩 증가시키는 Method를 100,000회 반복하여 두 번 실행하면 결과로 200000이 나와야 하지만, 아래 예제와 같이 다중 Thread에서는 실행할 때마다 원하지 않는 결과가 나올 수 있다.
여러 Thread가 동시에 수행되면서 Thread마다 같은 값의 Count 변수가 저장되는데, 이 변수를 증가시켜도 결국 같은 값을 저장하므로 한번 증가한 값과 같아지는 문제가 생기는 것이다.

class MyData
{
	public int Count { get; set; }
	
	public MyData()
	{
		this.Count = 0;
	}
	
	public void increment()
	{
		Count++;
	}
}

static void Main()
{
	MyData data = new MyData();
	
	Thread thread1 = new Thread(threadFunc);
	Thread thread2 = new Thread(threadFunc);
	
	thread1.Start(data);
	thread2.Start(data);
	
	thread1.Join();
	thread2.Join();
	
	Console.WriteLine(data.Number);
}

static void threadFunc(object inst)
{
	MyData data = inst as MyData;
	
	for (int i = 0; i < 100000; i++)
	{
		data.increment();
        }
}

출력결과 1 : 
173330

출력결과 2 : 
136522

출력결과 3 : 
98965

Monitor를 사용하면 하나의 Thread에서만 Count 변수에 접근하므로 200000이라는 정상적인 결과를 얻을 수 있다.

static void threadFunc(object inst)
{
	Mydata data = inst as MyData;
	
	for (int i = 0; i < 100000; i++)
	{
		Monitor.Enter(data);
		
		try
		{
			data.increment();
		}
		finally
		{
			Monitor.Exit(data);
		}
	}
}

출력결과 : 
200000

lock 키워드를 사용하면 try~finally와 Monitor의 역할을 대신 할 수 있다.

static void threadFunc(object inst)
{
	Mydata data = inst as MyData;
	
	for (int i = 0; i < 100000; i++)
	{
		lock (data)
		{
			data.increment();
		}
	}
}

출력결과 : 
200000

Interlocked

Interlocked에서는 여러 Thread가 동시에 하나의 변수를 연산하는 과정에서 발생할 수 있는 문제를 해결하기 위해 Atomic Operation을 제공한다. 특히 64bit 변수에서 이러한 문제가 발생할 수 있는데 Interlocked을 사용해 하나의 Thread로 한 번에 처리할 수 있다. 이 경우에 Monitor를 사용하여 Thread를 동기화할 수 있지만, Interlocked에서 제공하는 연산 기능으로 간단하게 구현할 수 있다.

static void threadFunc(object inst)
{
	long number = 0; // 8byte(64bit)
	
	Interlocked.Exchange(ref number, 3); // number = 5;
	Interlocked.Add(ref number, 5); // number += 5;
	Interlocked.Increment(ref number); // number++;
	Interlocked.Decrement(ref number); // number--;
}

ThreadPool

ThreadPool에서는 필요할 때마다 Thread를 꺼내 쓰고 다시 반환할 수 있는 기능을 제공한다. Thread를 따로 생성할 필요 없이 QueueUserWorkItem()을 사용하면 Thread가 자동으로 생성되고 지정한 Method가 할당되어 실행한다.

static void Main()
{
	Mydata data = new Mydata();
	
	ThreadPool.QueueUserWorkItem(threadFunc, data);
	ThreadPool.QueueUserWorkItem(threadFunc, data);
	
	Thread.Sleep(3000); // 3초 동안 실행 중지
	
	Console.WriteLine(data.Number);
}

EventWaitHandle

EventWaitHandle은 Signal과 Non-Signal의 상태 변화를 통해 Thread를 동기화하는 방법이다. WaitOn()에서 Non-Signal이 Signal 상태로 바뀔 때까지 Thread가 더 실행되지 못하게 하고 대기 상태로 만든다. 앞서 나온 Background Thread의 Join()과 유사한 역할을 한다.

static void Main()
{
	Console.WriteLine("Main Thread Start");
	
	EventWaitHandle ewh = new EventWaitHandle(false, EventResetMode.ManualReset); // Non-Signal 상태, 수동 Reset
	
	Thread thread = new Thread(threadFunc);
	thread.IsBackground = true;
	thread.Start(ewh);
	
	ewh.WaitOn(); // Signal 상태로 바뀔 때까지 대기
	
	Console.WriteLine("Main Thread End");
}

static void threadFunc(object state)
{
	EventWaitHandle ewh = state as EventWaitHandle;
	
	Console.WriteLine("Background Thread Start");
	
	Thread.Sleep(3000); // 3초 동안 실행 중지
	
	Console.WriteLine("Background Thread End");
	
	ewh.Set(); // Signal 상태로 전환
}

실행 결과 : 
Main Thread Start
Background Thread Start
Background Thread End
Main Thread End

Acync

Async Method

async 키워드를 사용하면 비동기 호출이 가능한 Async Method로 정의할 수 있다. Async Method에 비동기 호출을 적용하려면 해당 작업에 await 키워드를 사용한다. await를 사용한 작업이 완료될 때까지 이후 실행은 일시 중단하고 대기하게 된다.

public async void AwaitDownloadString()
{
	WebClient webClient = new WebClient();
	
	// 반환이 있는 Task<TResult>는 await를 사용하면 TResult 형식으로 반환된다.
	string text = await webClient.DownloadStringTaskAsync("http://www.novonetworks.com/");
	
	Console.WriteLine(text);
}

Task

Task는 Thread Pool을 사용하여 비동기 작업을 구현할 수 있도록 해준다.

반환 값이 없는 Task

Task task = new Tack(() => { Thread.Sleep(3000); });
task.Start();

Task를 생성할 필요 없이 Action Delegate를 전달하고 바로 작업을 시작할 수 있다.

Task task = Task.Factory.StartNew(() => { Thread.Sleep(3000); });

반환 값이 있는 Task

Task<int> task = new Task<int>(() =>
{
	Thread.Sleep(3000);
	return 3;
});

task.Start();

Console.WriteLine(task.Result); // 출력 결과 : 3

Task<TResult>를 생성할 필요 없이 Func Delegate를 전달하고 바로 작업을 시작할 수 있다.

Task<int> task = Task.Factory.StartNew<int>(() =>
{
	Thread.Sleep(3000);
	return 3;
});


Console.WriteLine(task.Result); // 출력 결과 : 3

Async Method가 아닌 경우의 비동기 처리

Async Method가 아닌 Method의 반환 형식을 **Task**로 바꾸면 await를 이용해 비동기 호출을 적용할 수 있다.

public async void AwaitDownloadString()
{
	string text = await MyDownloadString("http://www.novonetworks.com/");
	
	Console.WriteLine(text);
}

public Task<string> MyDownloadString(string uri)
{
	WebClient webClient = new WebClient();
	
	return Task.Factory.StartNew(() =>
	{
		return webClient.DownloadString(uri);
	});
}

비동기 호출의 병렬 처리

**Task.WaitAll()**는 여러 작업을 병렬로 수행하는 Method이다. **Task.WaitAll()**에 await를 사용하면 여러 작업을 동시에 비동기 호출로 처리할 수 있다.

ThreeSeconds()와 FiveSeconds()를 각각 수행하려면 총 '8초'가 걸리지만, 두 작업을 병렬로 수행하면 두 작업의 최대 소요시간인 '5초'에 완료하는 것을 확인할 수 있다.

static void Main()
{
	DoAsyncTask();
	
	Console.WriteLine("Main Thread End");
}

static async void DoAsyncTask()
{
	Stopwatch stopWatch = new Stopwatch();
	
	var task1 = ThreeSeconds();
	var task2 = FiveSeconds();
	
	stopWatch.Start();
	await Task.WhenAll(task1, task2);
	stopWatch.Stop();
	
	TimeSpan timeSpan = stopWatch.Elapsed;
	
	Console.WriteLine("Total : " + (task1.Result + task2.Result));
	Console.WriteLine("Result : " + timeSpan.Seconds);
}

static Task<int> ThreeSeconds()
{
	return Task.Factory.StartNew(() =>
	{
		Thread.Sleep(3000); // 3초
		return 3;
	});
}

static Task<int> FiveSeconds()
{
	return Task.Factory.StartNew(() =>
	{
		Thread.Sleep(5000); // 5초
		return 5;
	});
}

출력 결과 : 
Main Thread End
Total : 8
Result : 5

예제 분석

SendRecvCommandTimeoutAsync()

public async Task<Tuple<bool, int>> SendRecvCommandTimeoutAsync(/* 매개변수 */) { /* 내용 */ }

async를 사용했으므로 비동기 호출이 가능한 Method이다.
bool과 int 형식의 값을 저장한 Tuple 객체를 가진 Task를 반환한다.

Task<int> cmdTask = SendRecvCommandAsync(client, ipAddr, cmd, ctag, sessionID);
Task timeoutTask = Task.Delay(timeoutMillis);

ENC로 부터 PDU를 전송받아 calla_id로 변환된 값을 반환하는 Task인 SendRecvCommandAsync()를 cmdTask에 저장한다.
현재 Method에서 매개변수로 받은 값인 timeoutMillis(기본 값:10000) 만큼 Delay 하는 Task를 timeoutTask에 저장한다.

result = await Task.Factory.ContinueWhenAny<bool>(new Task[] { cmdTask, timeoutTask }, (completedTask) => { /* 내용 */ });

cmdTask와 timeoutTask를 동시에 실행하여 먼저 끝난 Task를 Func Delegate의 매개변수인 completedTask에 넘긴 후 Task를 실행한다.
await를 사용했으므로 이 모든 Task는 비동기로 처리되고, Task의 반환 값은 result에 저장한다.
Task.Factory.ContinueWhenAny()와 관련된 내용과 예제를 설명 하단에 추가하였다.

if (completedTask == timeoutTask)
{
	return false;
}
else if
{
	/* 생략 */
}
else
{
	return false;
}

completedTask와 timeoutTask가 같다는 것은 cmdTask을 작업하는 데 제한시간을 초과했다는 것이므로 false를 반환한다. 이외 경우도 시간 초과로 보고 false를 반환한다.

else if (completedTask == cmdTask)
{
	calla_id = cmdTask.Result;
	
	int isValid = Calla.valid(calla_id);
	if (isValid != 1)
	{
		Calla.close(calla_id);
	}
	
	return (isValid == 1);
}

completedTask와 cmdTask가 같다는 것은 제한시간 내에 cmdTask에서 calla_id를 반환했다는 것이다.
calla_id가 유효한 ID인지를 확인한다. Invalid(-1)하면 Close하고, Valid(1)하면 true를 반환한다.

return new Tuple<bool, int>(result, calla_id);

Task.Factory.ContinueWhenAny()에서 반환된 result와 cmdTask에서 반환된 calla_id 변수를 Tuple에 저장 후 객체를 생성하여 최종적으로 반환한다.

Task.Factory.ContinueWhenAny()

예제를 살펴보면 첫 번째 매개변수의 Task 배열 중 먼저 끝난 Task가 두 번째 Func Delegate의 매개변수로 전달되는 것을 알 수 있다.

public async void asyncTest()
{
	var task1 = Task.Factory.StartNew(() =>
	{
		Thread.Sleep(5000);
		return 1;
	});

	var task2 = Task.Factory.StartNew(() =>
	{
		Thread.Sleep(2000);
		return 2;
	});

	var task3 = Task.Factory.StartNew(() =>
	{
		Thread.Sleep(3000);
		return 3;
	});
	
	int result = await Task.Factory.ContinueWhenAny<int>(new[] { task1, task2, task3 }, (completedTask) =>
	{
		int id = -1;

		if (completedTask.Equals(task1))
		{
			id = task1.Result;
		}
		else if (completedTask.Equals(task2))
		{
			id = task2.Result;
		}
		else if (completedTask.Equals(task3))
		{
			id = task3.Result;
		}

		return id;
	});
	
	Console.WriteLine(result); // 실행 결과 : 2 (task2의 반환 값)
}

SendRecvCommandAsync()

// App.Const.cs

public static class AppConst
{
	// PDU Header
	public enum PduHeader : int
	{
		signature = 0,	// [4] SIGNATURE	$HI$"
		version = 4,	// [1] VERSION		1
		type = 5,	// [1] VERSION		reqTL1(0), rspCalla(1), rspTL1(2)
		checksun = 6,	// [2] CHECKSUM	sum(body)
		session = 8,	// [4] SESSION ID
		ctag = 12,	// [4] USER TAG
		length = 16,// [4] BODY LENGTH
		Size = 20
	};
	
	public const int PduDefaultVersion = 1;
	public const int PduRequestTL1 = 0;
	public const int PduResponseCalla = 1;
	public const int PduSessionFreePass = 999999999;
	public const int PduSessionInitial = 0;
	
	// Calla Const
	public const int CallaHeaderSize = 24;
	public const int CallaOneRowSize = 1024;
	
	public const int CallaBufferNumber = 50;
	public const int CallaBufferSize = (150 * CallaOneRowSize) + CallaHeaderSize;
	
	// TCP PORT
	public const int EncTcpPort = 3300;
	
	/* 생략 */
}

// CRC.CheckSum16() Method 수정

ushort CheckSum16(byte[] ptr, int len)
{
	uint sum;
	ushort chsum;
	byte over;
	
	byte index = 0; // 추가
	
	sum = 0;
	for (; len > 1; len -= 2)
	{
		sum += ptr[index];
		index++;
	}
	
	if (len == 1)
	{
		ptr[index] = (byte)(ptr[index] & 0xff00);
		sum += (uint)(ptr[index] >> 8);
	}
	
	over = (byte)(sum >> 16);
	chsum = (ushort)(~(over + (ushort)sum));
	
	return chsum;
}

// SendRecvCommandAsync()에 필요한 임의의 변수들 선언

TcpClient client = new TcpClient();
string ipAddr = "192.168.0.100"; 
string cmd = "TEST";
int ctag = 1234;
int sessionID = AppConst.PduSessionFreePass; // 999999999

Task<int> cmdTask = SendRecvCommandAsync(client, ipAddr, cmd, ctag, sessionID);

public async Task<int> SendRecvCommandAsync(TcpClient client, string ipAddr, string cmd, int ctag, int sessionID) { /* 내용 */ }

async를 사용했으므로 비동기 호출이 가능한 Method이다.
int 형식의 값을 가진 Task를 반환한다.

string myCmd = "%%" + cmd;
int calla_id = -1;

전송할 명령어 앞에 기본적으로 "%%"를 추가한다. (ex. %%TEST)
calla_id를 –1로 초기화해준다. (Invalid = -1)

await client.ConnectAsync(ipAddr, EncClientManager.Port);

IP(ipAddr)와 Port(EncClientManager.Port)에 해당하는 TCP Sever(ENC)에 Client를 연결한다.
TCP Server와 연결하는 과정에서 대기시간이 길어지게 되면 병목이 발생하여 프로그램을 실행하는 동안에 문제가 발생할 수 있다. await를 사용해 비동기 처리를 함으로써 이러한 문제를 방지할 수 있다.

using (var stream = client.GetStream())
{
	/* 내용 */
}

연결된 Client에서 데이터를 송수신하는 데 사용되는 Stream을 가져온다.
using 키워드를 사용하여 코드가 끝나거나 예외가 발생했을 때 Dispose()를 호출하여 객체의 할당을 자동으로 해제한다.

byte[] cmdBytes = Encoding.UTF8.GetBytes(myCmd);
int paddedCmdLen = cmdBytes.Length;
if ((paddedCmdLen % 2) == 1)
	paddedCmdLen++;

전송할 명령어를 byte로 Encoding하여 cmdBytes에 저장한다.
cmdBytes의 길이가 홀수이면 명령어의 크기를 1 증가시켜 짝수로 만들어 준다. (Alignment)

byte[] paddedCmdBytes = new byte[paddedCmdLen];
Array.Clear(paddedCmdBytes, 0, paddedCmdBytes.Length);
Array.Copy(cmdBytes, paddedCmdBytes, cmdBytes.Length);

명령어 크기만큼의 byte 배열(paddedCmdBytes)을 생성한다.
paddedCmdBytes 배열의 모든 값을 기본 값인 0으로 초기화한다.
cmdBytes 배열 전체를 paddedCmdBytes에 복사한다. paddedCmdBytes에는 전송할 Payload를 저장하고 있다.

byte[] sendBuffer = new byte[PDU_HDR_LEN + paddedCmdBytes.Length];

PDU의 Header와 paddedCmdBytes의 길이를 더한 만큼의 byte 배열(sendBuffer)을 생성한다. sendBuffer는 ENC로 전송될 PDU가 될 것이다.

Array.Copy(PduSignature, sendBuffer, PduSignature.Length);

정의된 PduSignature의 값을 sendBuffer의 첫 번째 요소부터 복사한다.

sendBuffer[(int)AppConst.PduHeader.version] = AppConst.PduDefaultVersion;

PduDefaultVersion를 sendBuffer의 4번 요소에 저장한다.

sendBuffer[(int)AppConst.PduHeader.type] = AppConst.PduRequestTL1;

PduRequestTL1를 sendBuffer의 5번 요소에 저장한다.

ushort checksum = CRC.CheckSum16(paddedCmdBytes, paddedCmdBytes.Length);
byte[] checksumBytes = BitConverter.GetBytes(checksum);
Array.Reverse(checksumBytes);
Array.Copy(checksumBytes, 0, sendBuffer, (int)AppConst.PduHeader.checksun, checksumBytes.Length);

paddedCmdBytes에 대한 CRC 값을 checksum을 저장한다.
checksum을 byte 로 Convert 해서 checksumBytes에 저장한다.
checksumBytesd를 뒤집어서 sendBuffer의 6~7번 요소에 복사한다. (Endian 변환)

byte[] sessBytes = BitConverter.GetBytes(sessionID);
Array.Reverse(sessBytes);
Array.Copy(sessBytes, 0, sendBuffer, (int)AppConst.PduHeader.session, sessBytes.Length);

sessionID를 byte로 Convert 해서 sessBytes에 저장한다.
sessBytes를 뒤집어서 sendBuffer의 8~11번 요소에 복사한다.

byte[] ctagBytes = BitConverter.GetBytes(ctag);
Array.Reverse(ctagBytes);
Array.Copy(ctagBytes, 0, sendBuffer, (int)AppConst.PduHeader.ctag, ctagBytes.Length);

ctag를 byte로 Convert 해서 ctagBytes에 저장한다.
ctagBytes를 뒤집어서 sendBuffer의 12~15번 요소에 복사한다.

byte[] lengthBytes = BitConverter.GetBytes(paddedCmdBytes.Length);
Array.Reverse(lengthBytes);
Array.Copy(lengthBytes, 0, sendBuffer, (int)AppConst.PduHeader.length, lengthBytes.Length);

paddedCmdBytes의 길이를 byte로 Convert 해서 lengthBytes에 저장한다.
lengthBytes를 뒤집어서 sendBuffer의 16~19번 요소에 복사한다.

Array.Copy(paddedCmdBytes, 0, sendBuffer, (int)AppConst.PduHeader.Size, paddedCmdBytes.Length);

paddedCmdBytes를 sendBuffer의 20번 요소부터 복사한다.
ENC로 전송될 PDU의 정보가 sendBuffer에 모두 저장되었다.

await stream.WriteAsync(sendBuffer, 0, sendBuffer.Length);

현재 Stream에 해당하는 ENC로 PDU(sendBuffer)를 비동기 처리로 전송한다.

byte[] recvBuffer = new byte[AppConst.CallaBufferSize];

전송받을 Calla PDU 크기만큼의 byte 배열(recvBuffer)을 생성한다.

int bytes;

bytes = await stream.ReadAsync(recvBuffer, 0, PDU_HDR_LEN);
if (bytes == 0)
{
	Trace.TraceWarning(/* 생략 */);
	return -1;
 }

현재 Stream에 해당하는 ENC로부터 Header 길이만큼의 byte 데이터를 비동기 처리로 전송받는다.
전송받은 byte 수가 0이면 오류로 판단하여 -1을 반환한다.

bool isValidSignature = true;
for (int i = 0; i < PduSignature.Length; i++)
{
	if (recvBuffer[i] != PduSignature[i])
	{
		isValidSignature = false;
		break;
	}
}

if (!isValidSignature)
{
	Trace.TraceWarning(/* 생략 */);
	return -1;
}

지정된 Signature와 recvBuffer의 Signature가 같은지 확인하고, 다르면 오류로 판단하여 -1을 반환한다.

byte version;			// Version
byte type;			// Palyload Type
ushort recvChecksum;		// Checksum
int sessID;			// Session ID
int userTag;			// User Tag
int bodyLength;			// Body Length

PDU의 Header 정보를 저장할 각각의 변수를 선언한다.

version = recvBuffer[(int)AppConst.PduHeader.version];
type = recvBuffer[(int)AppConst.PduHeader.type];

if (type != AppConst.PduResponseCalla)
{
	Trace.TraceWarning(/* 생략 */);
	return -1;
}

recvBuffer의 4번 요소의 값을 version에 저장한다.
recvBuffer의 5번 요소의 값을 type에 저장한다.
type이 1(Calla Response)이 아니면 오류로 판단하여 -1을 반환한다.

Array.Reverse(recvBuffer, (int)AppConst.PduHeader.checksun, 2);
recvChecksum = BitConverter.ToUInt16(recvBuffer, (int)AppConst.PduHeader.checksun);

recvBuffer의 6~7번 요소의 값을 뒤집은 후 UInt16으로 Convert 해서 recvChecksum에 저장한다.

Array.Reverse(recvBuffer, (int)AppConst.PduHeader.session, 4);
sessID = BitConverter.ToInt32(recvBuffer, (int)AppConst.PduHeader.session);

recvBuffer의 8~11번 요소의 값을 뒤집은 후 Int32로 Convert 해서 sessID에 저장한다.

Array.Reverse(recvBuffer, (int)AppConst.PduHeader.ctag, 4);
userTag = BitConverter.ToInt32(recvBuffer, (int)AppConst.PduHeader.ctag);

if (userTag != ctag)
{
	Trace.TraceWarning(/* 생략 */);
	return -1;
}

recvBuffer의 12~15번 요소의 값을 뒤집은 후 Int32로 Convert 해서 userTag에 저장한다.
ENC로 전송했던 ctag와 userTag가 다르면 오류로 판단하여 -1을 반환한다.

Array.Reverse(recvBuffer, (int)AppConst.PduHeader.length, 4);
bodyLength = BitConverter.ToInt32(recvBuffer, (int)AppConst.PduHeader.length);

recvBuffer의 16~19번 요소의 값을 뒤집은 후 Int32로 Convert 해서 bodyLength에 저장한다.

int remainLength = bodyLength;
int destIndex = 0;

while (remainLength > 0)
{
	int readAmount = await stream.ReadAsync(recvBuffer, destIndex, remainLength);
	if (readAmount == 0)
	{
		Trace.TraceWarning(/* 생략 */);
		return -1;
	}
	remainLength -= readAmount;
	destIndex += readAmount;
}

전송받은 후 남은 데이터의 길이를 확인할 변수인 remainLength에 Payload의 길이를 저장하고, 현재까지 전송받은 데이터의 길이를 확인할 변수인 destIndex에 0을 초기화 해준다.
이 전에 ReadAsync()를 통해 Header 부분은 전송을 받았으므로, Body 부분부터 전송을 받을 것이다. 전송받은 index(destIndex)부터 남은 데이터의 길이(remainLength)만큼의 byte 데이터를 비동기 처리로 전송받는다.
한 번에 많은 데이터를 전송받지 못 하므로 readAmount에 전송받은 데이터의 길이를 저장하여 남은 길이를 확인할 것이다.
남은 데이터의 길이(remainLength)가 있음에도 전송받은 데이터(readAmount)가 없으면 오류로 판단하여 -1을 반환한다.
남은 데이터의 길이(remainLength)에서 전송받은 데이터의 길이(readAmount)를 제외한다. 또한, 다음에 전송받은 데이터를 이어서 저장하기 위해 recvBuffer의 index를 확인하는 변수인 destIndex에 전송받은 데이터의 길이(readAmount)를 더해준다.

if (!Calla.SafeExtract(recvBuffer, bodyLength, out calla_id))
{
	;   // TODO.... PDU overflow or PDU crashed...
}

전송받은 Payload를 Parsing해서 변환된 값을 calla_id에 저장한다.

bool isValid = (Calla.valid(calla_id) == 1);

calla_id가 유효한 ID인지를 확인한다. (Valid = 1, Invalid = -1)

return calla_id;

SendRecvCommandAsync()는 최종적으로 calla_id를 반환한다.