2
\$\begingroup\$

I've recently been dipping my toes in to async code so I'm not 100% sure if there's a better way to do this, or if I'm going to run into problems down the road.

I plan on using the following class in Unity. I'll be capturing a minimal amount of game state data every frame (between 25-100 bytes). I add the data to a queue and when it reaches around 1kb, I give it to my client object to serialize and send it to a server via tcp. I've called each data holding queue a 'chunk'. (Packet probably would have been better)

I need to minimize blocking of Unity's main thread, I don't want there to be any opportunity for my client tasks to block the game thread so this class is supposed to act as a layer between the two. Recorder is probably a bad name for it.

Everything works right now, I'm just unsure if theres a better way to (nearly) asynchronously pass data through queues for processing between threads.

This class will be used like so:

Recorder recorder = new Recorder();
recorder.StartRecording(client, 30); //30 frames per chunk

...

//every frame
recorder.Record(frameData);

...

recorder.StopRecording();

frameData is any serializable class or struct. I have my own

public class Recorder<T> {

    private BlockingCollection<Queue<T>> dataQueue;
    private BlockingCollection<T> receivingQueue;
    private int chunkSize;
    private CancellationTokenSource cts;
    private Client client;


    public Recorder(Client _client, int _chunkSize) {
        dataQueue = new BlockingCollection<Queue<T>>();
        receivingQueue = new BlockingCollection<T>();
        client = _client;
        chunkSize = _chunkSize;
        cts = new CancellationTokenSource();

    }

    public void StartRecording() {
        CancellationToken token = cts.Token;
        Task.Factory.StartNew(() => QueueFillerTask(token), token);
        Task.Factory.StartNew(() => DataSendingTask(token), token);
    }

    public void StopRecording() {
        cts.Cancel();
    }

    public void Record(T obj) {
        receivingQueue.Add(obj);
    }

    private void QueueFillerTask(CancellationToken token) {
        while (true) {

            Queue<T> chunk = new Queue<T>();

            try {
                while (chunk.Count < chunkSize) {
                    T obj = receivingQueue.Take(token);
                    chunk.Enqueue(obj);
                }
                dataQueue.Add(chunk);
                Logger.Log("Chunk added");
            } catch (System.OperationCanceledException e) {
                if (chunk.Count > 0) { //Put the incomplete chunk in the dataqueue.
                    dataQueue.Add(chunk);
                }
                Logger.Log("Recording ended");
                break;
            }
        }
    }

    private void DataSendingTask(CancellationToken token) {
        try {
            while (true) {
                Queue<T> data = dataQueue.Take(token);
                client.PackAndShip(data);
            }

        } catch (System.OperationCanceledException e) {
            bool moreData;
            do {
                Queue<T> remainingItem;
                moreData = dataQueue.TryTake(out remainingItem);
                if (moreData) {
                    client.PackAndShip(moreData);
                }
            } while (moreData);
            Logger.Log("Sending ended");
        }
    }

}
\$\endgroup\$
  • \$\begingroup\$ Could you explain how the two queues work together? I'm having a hard time figuring this out. \$\endgroup\$ – t3chb0t Oct 29 '18 at 8:37
  • \$\begingroup\$ Record(obj) adds a message to the receiving queue, \$\endgroup\$ – itsDrew Oct 29 '18 at 12:15
  • \$\begingroup\$ ok, I already know that ;-] I had hoped you would explain why you are doing this etc. since this is not something one sees on an everyday basis - moving items from one queue to another one. \$\endgroup\$ – t3chb0t Oct 29 '18 at 12:17
  • \$\begingroup\$ Record(obj) adds an object to receiving queue. Objects go from the receiving queue to another container (which happens to also be a queue). Once this container fills to the desired size, it is ready to be sent (it is put in to the data queue). Objects from the data queue are pulled out by the client.packandship method. \$\endgroup\$ – itsDrew Oct 29 '18 at 12:21
  • \$\begingroup\$ Maybe it's overkill? I am receiving individual objects, filling a ~1kb box with them and sending the box. (Everything needs to be reliable and ordered, so this is faster I think than sending each object individually). The longest running operation will be the packandship method. If there is a latency spike, that will block. The chunk (container) I am using is a queue, but it could be any collection. I just happen to be using a queue since thats the data structure I need on the other side. \$\endgroup\$ – itsDrew Oct 29 '18 at 12:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.