[C++] JobQueue (1 / 3)

멀티 스레드 환경이기 때문에 JobQueue의 중요성이 두드러진다.

이에 대해 알아보자.

1. JobQueue가 왜 필요한데?

스레드끼리의 경합을 예방해 더 효율적인 처리를 가능케 하기 위함이다.

밥집을 갔다고 생각해 보자.

손님들이 주문서를 작성하는 것이 아닌, 바로 주방에 와서 이것저것 주문한다. 주방엔 주방장 단 한 명.

주문이 오는 대로 주방장은 만들기 시작하고, 주방장은 한 명뿐이기 때문에

주방장이 내 주문을 만들게 하기 위한 경합이 생긴다.

먼저 주문하려고 했어도 경합에 실패에 순서가 밀릴 수도 있다.

여러 개의 음식이 갖춰져야 하는 상황이라고 했을 때 경합에 계속 실패한다면 문제가 된다.

손님들은 주방 앞에서 내 주문이 완료될 때까지 기다려야 하니 답답하다.

손님 나름대로 다른 일을 하지 못하게 된 셈.

따라서 경합을 없애고 순서를 지키도록 하고, 손님은 주문서만 던져주고 다른 일을 할 수 있게 하기 위해,

JobQueue가 필요하게 됐다는 것이다.

2. 클래스 작성

JobQueue의 역할을 할 클래스를 만들어 보자.

#pragma once

class IJob
{
public:
    virtual void Execute() { }
};

using JobRef = shared_ptr<IJob>;

class JobQueue
{
public:
    void Push(JobRef job)
    {
        WRITE_LOCK;
        _jobs.push(job);        
    }

    JobRef Pop()
    {
        WRITE_LOCK;
        if (_jobs.empty())
            return nullptr;

        JobRef ret = _jobs.front();
        _jobs.pop();
        return ret;
    }

private:
    USE_LOCK;
    queue<JobRef> _jobs;
};

• IJob
  • 큐에 들어갈 잡들의 인터페이스
  • 상속받을 클래스에 Execute()를 구현
• JobQueue
  • 큐의 역할을 수행할 클래스

큐의 각 함수 및 _jobs에 락이 걸려있는 것을 볼 수 있다.

Thread-Safe 한 처리를 위해 락이 필요하다.

예시로 사용할 Job은 아래와 같이 만들었다.

class HealJob : public IJob
{
public:
    virtual void Execute() override
    {
        cout << _target << "에게 " << _healValue << " 만큼 힐" << endl;
    }

public:
    uint64 _target = 0;
    uint32 _healValue = 0;
};

2. 기존 클래스 대응

Room 클래스에 JobQueue를 넣어보자.

// Room.h

#pragma once
#include "Job.h"

class Room
{
    friend class EnterJob;
    friend class LeaveJob;
    friend class BroadcastJob;

private:
    void Enter(PlayerRef player);
    void Leave(PlayerRef player);
    void Broadcast(SendBufferRef sendBuffer);

public:
    void PushJob(JobRef job) { _jobs.Push(job); }
    void FlushJob();

private:
    map<uint64, PlayerRef> _players;

    JobQueue _jobs;
};


// Room.cpp

#include "pch.h"
#include "Room.h"
#include "Player.h"
#include "GameSession.h"

Room GRoom;

void Room::Enter(PlayerRef player)
{
    _players[player->playerId] = player;
}

void Room::Leave(PlayerRef player)
{
    _players.erase(player->playerId);
}

void Room::Broadcast(SendBufferRef sendBuffer)
{
    for (auto& p : _players)
    {
        p.second->ownerSession->Send(sendBuffer);
    }
}

void Room::FlushJob()
{
    while (true)
    {
        JobRef job = _jobs.Pop();
        if (job == nullptr)
            break;

        job->Execute();
    }
}

Room의 기본적인 기능을 수행할 수 있게 기본적인 틀을 잡았다.

private인 함수를 각 Job 클래스에서 사용할 수 있도록 friend 클래스로 지정했다.

각 클래스들은 아래와 같다.

// Room Jobs
class EnterJob : public IJob
{
public:
    EnterJob(Room& room, PlayerRef player) : _room(room), _player(player)
    {
    }

    virtual void Execute() override
    {
        _room.Enter(_player);
    }

public:
    Room& _room;
    PlayerRef _player;
};

class LeaveJob : public IJob
{
public:
    LeaveJob(Room& room, PlayerRef player) : _room(room), _player(player)
    {
    }

    virtual void Execute() override
    {
        _room.Leave(_player);
    }

public:
    Room& _room;
    PlayerRef _player;
};

class BroadcastJob : public IJob
{
public:
    BroadcastJob(Room& room, SendBufferRef sendBuffer) : _room(room), _sendBuffer(sendBuffer)
    {
    }

    virtual void Execute() override
    {
        _room.Broadcast(_sendBuffer);
    }

public:
    Room& _room;
    SendBufferRef _sendBuffer;
};

여기서 확인할 수 있는 가장 큰 특징은 「락이 걸려있지 않다」는 것이다.

락을 걸지 않으면 Thread-Safe 하지 않다고 생각할 수 있다.

이게 문제 되지 않는 이유는 「이미 Job에 락이 다 걸려있기 때문」이다.

큐의 모든 Push와 Pop엔 락이 걸려 있다.

단순히 어떤 일을 처리해 달라고 요청을 큐에 보내고 쌓아둘 뿐이고,

큐가 알아서 락을 건 후 하나씩 빼서 처리할 것이기 때문에 요청할 작업에 일일이 락을 걸 필요가 없는 것이다.

그리고 새로운 잡이 생길 때마다 클래스를 늘려야 하는 것은 매우 비효율적인 작업인 것 같다.

이는 추후의 개선사항으로 남겨두자.

3. 테스트

서버에 아래와 같은 테스트용 잡을 추가 했다.

// [일감 의뢰 내용] : 1번 유저한테 10만큼 힐을 줘라!
// 행동 : Heal
// 인자 : 1번 유저, 10이라는 힐량
HealJob healJob;
healJob._target = 1;
healJob._healValue = 10;

// 나~중에
healJob.Execute();

그리고 클라이언트로부터 오는 일감을 처리하기 위한 부분도 작성한다.

while (true)
{
    GRoom.FlushJob();
    this_thread::sleep_for(1ms);
}

또한 빌드하면서 기존 함수를 사용하는 부분들을 다 Job을 사용하게끔 수정해 준다.

실행해 보면 아래와 같다.

처리가 잘 됨을 알 수 있다.

4. 생각해 볼 점

bool Handle_C_ENTER_GAME(PacketSessionRef& session, Protocol::C_ENTER_GAME& pkt)
{
    GameSessionRef gameSession = static_pointer_cast<GameSession>(session);

    uint64 index = pkt.playerindex();
    // TODO : Validation

    PlayerRef player = gameSession->_players[index]; // READ_ONLY?

    GRoom.PushJob(MakeShared<EnterJob>(GRoom, player));

    Protocol::S_ENTER_GAME enterGamePkt;
    enterGamePkt.set_success(true);
    auto sendBuffer = ClientPacketHandler::MakeSendBuffer(enterGamePkt);
    player->ownerSession->Send(sendBuffer);

    return true;
}

위 함수를 보자.

실행하면서 EnterJob을 이미 큐에 넘겨줬다.

하지만 잡은 바로 실행되지 않기 때문에 처리가 어떻게 됐는지도 모르는 상태에서 클라에 패킷을 넘겨주는 것은

바람직한 처리 방법은 아닌 것 같다.

이를 해결할 방법을 찾을 필요가 있다.

A. Functor

잡 클래스를 계속 찍어내는 것이 아닌 Fucntor(함수자)를 활용하는 방법이 있다.

위 방법이 1세대라면 이건 2세대 정도.

여기선 실제로 작업을 수행할 함수와 함수자를 활용하여 아까와 같은 비효율적인 반복을 방지한다.

// 함수자 (Functor)
class IJob
{
public:
    virtual void Execute() { }
};

template<typename Ret, typename... Args>
class FuncJob : public IJob
{
    // FuncType은 Ret을 뱉어주는 함수인데, 인자로는 여러개를 받을 수 있다.
    using FuncType = Ret(*)(Args...);

public:
    FuncJob(FuncType func, Args... args) : _func(func), _tuple(args...)
    {

    }

    // 잡으로 던져준다는 것 자체가
    // 당장 리턴값이 필요하다는 뜻이 아님.
    virtual void Execute() override
    {
        std::apply(_func, _tuple); // C++17
    }

private:
    FuncType _func;
    std::tuple<Args...> _tuple;
};

튜플의 형태로 인자를 넘겨주기 때문에 C++17에서 추가된 apply()를 활용할 수 있다.

이를 사용하면 하나의 템플릿 함수자로 인자의 개수에 관계없이 함수를 호출할 수 있게 된다.

하지만 위의 형식으로는 어떤 클래스에 있는 멤버 함수를 호출할 수는 없다. 아예 호출 규약이 다르다.

그 함수가 static이라면 호출할 수 있겠지만 모든 멤버 함수가 static일 수는 없으니,

멤버 함수 호출 규약인 __thiscall에 따라 this 포인터를 넘겨주어야 한다.

template<typename T, typename Ret, typename... Args>
class MemberJob : public IJob
{
    // 어떤 클래스 안에 포함된 함수를 나타냄.
    using FuncType = Ret(T::*)(Args...);

public:
    MemberJob(T* obj, FuncType func, Args... args) : _obj(obj), _func(func), _tuple(args...)
    {

    }

    virtual void Execute() override
    {
        std::apply([this](Args... args) { return (_obj->*_func)(args...); }, _tuple);
    }

// _obj 등은 나중에 사용
private:
    T*            _obj;
    FuncType    _func;
    std::tuple<Args...> _tuple;
};

위와 같이 람다식을 사용해 this를 넘겨주며 멤버 함수를 호출할 수 있다.

하지만 std::apply()는 C++17에서 사용 가능한 기능이다.

그럼 이전엔 어떤 형태로 이를 구현했을지 궁금하다.

A-1. C++11에서의 구현

일단 전체 코드를 보자.

// C++11 apply
template<int... Remains>
struct seq
{};

template<int N, int... Remains>
struct gen_seq : gen_seq<N-1, N-1, Remains...>
{};

// 맨 앞의 값이 0일 때 사용됨
template<int... Remains>
struct gen_seq<0, Remains...> : seq<Remains...>
{};

// 그냥 함수를 받음
template<typename Ret, typename... Args>
void xapply(Ret(*func)(Args...), std::tuple<Args...>& tup)
{
    xapply_helper(func, gen_seq<sizeof...(Args)>(), tup);
}

template<typename F, typename... Args, int... ls>
void xapply_helper(F func, seq<ls...>, std::tuple<Args...>& tup)
{
    (func)(std::get<ls>(tup)...);
}

// 클래스 안의 함수를 사용하기 위함
template<typename T, typename Ret, typename... Args>
void xapply(T* obj, Ret(T::*func)(Args...), std::tuple<Args...>& tup)
{
    xapply_helper(obj, func, gen_seq<sizeof...(Args)>(), tup);
}

template<typename T, typename F, typename... Args, int... ls>
void xapply_helper(T* obj, F func, seq<ls...>, std::tuple<Args...>& tup)
{
    (obj->*func)(std::get<ls>(tup)...);
}

솔직히 템플릿 흑마법들을 제대로 이해하기엔 쉽지 않은 것 같다.

각 부분은 간략히 보자면 아래와 같다.

• seq, gen_seq
  • seq : 함수에 넘겨줄 인자들을 갖고 있는 시퀀스
  • gen_seq : 위의 시퀀스를 만들 Generator
• xapply
  • 코드에서 std::apply()의 역할을 할 함수
• xapply_helper
  • xapply의 실제 처리가 이루어지는 부분

특히 궁금했던 부분은 seq를 만드는 부분이었는데, 그 흐름은 아래와 같다.

1. gen_seq<3>()이 있다고 가정하자.
2. <N-1, N-1>로 인해 gen_seq<2, 2>로 변한다.
3. 이를 반복하므로 gen_seq<1, 1, 2>로,
4. gen_seq<0, 0, 1, 2>로 변했고, 맨 앞이 0이기 때문에 gen_seq<0, Remains...>가 호출된다.
5. Remains만 갖게 되므로, 최종적으로 seq<0, 1, 2>가 된다.

템플릿에서 튜플의 인자를 꺼내오기 위해 get<>을 사용하므로,

배열처럼 꺼낼 수 있게 하기 위해 인덱스를 만드는 과정이었다.

이를 활용하면 기존의 std::apply()를 사용하는 부분은 이렇게 변한다.

virtual void Execute() override
{
    //std::apply(_func, _tuple); // C++17
    xapply(_func, _tuple);
}


template<typename T, typename Ret, typename... Args>
class MemberJob : public IJob
{
    // 어떤 클래스 안에 포함된 함수를 나타냄.
    using FuncType = Ret(T::*)(Args...);

public:
    MemberJob(T* obj, FuncType func, Args... args) : _obj(obj), _func(func), _tuple(args...)
    {

    }

    virtual void Execute() override
    {
        //std::apply([this](Args... args) { return (_obj->*_func)(args...); }, _tuple);        
        xapply(_obj, _func, _tuple);
    }

private:
    T*            _obj;
    FuncType    _func;
    std::tuple<Args...> _tuple;
};

"계속 개선되는 Modern C++을 공부하지 않으면 안 되겠구나" 하는 감정이 들었다.

A-2. 테스트

아래와 같은 함수를 준비했다.

void HealByValue(int64 target, int32 value)
{
    cout << target << "에게 " << value << " 만큼 힐" << endl;
}

class Knight
{
public:
    void HealMe(int32 value)
    {
        cout << "HealMe! " << value << endl;
    }
};

이를 활용하는 것은 아래와 같다.

// TEST JOB
{
    FuncJob<void, int64, int32> job(HealByValue, 100, 10);
    job.Execute();
}
{
    Knight k1;
    MemberJob job2(&k1, &Knight::HealMe, 10);
    job2.Execute();
}

실행을 확인해 보자.

의도한 대로 잘 동작함을 확인할 수 있다.