3

I'm using SQS as a queue for video encoding and want to ensure that only a single encoding is performed per video.

SQS works fine in that when a message is queued, it will only be received by a single thread. However, it's possible that multiple messages could be sent to the queue for the same video/encoding, meaning the message content would be the same for the particular 'encoding' queue.

Is there anyway to de-duplicate to ensure that for a specific queue, that the messages in the queue or received from a queue, are unique?

One option I thought would be to create a new queue for each encoding type, as the message is sent. So the queue could be named something like encoding-video-id, which would only have a single message and I could check to ensure that the queue does not yet exist. The only "issue" is that there could be 1000's to 10's of thousands of these queues created.

6
  • So what could cause you to enqueue the same message multiple times?
    – Mike Brant
    Dec 1, 2015 at 17:21
  • The use case is that users can submit 'encode' which queues the video, in edge cases it's possible for it to be hit multiple times, which would result in multiple messages.
    – dzm
    Dec 1, 2015 at 17:33
  • Just noticed you can create "unlimited" queues in sqs, so possibly the option above could work.
    – dzm
    Dec 1, 2015 at 17:35
  • Even without the possibility of a user queuing a duplicate task, SQS itself does not guarantee "exactly once" delivery of a message. It guarantees "at least once", so SQS itself can deliver duplicate messages. I think the answers to these questions are relevant to your issue: stackoverflow.com/questions/32386877/… and stackoverflow.com/questions/13484845/…
    – Mark B
    Dec 1, 2015 at 18:29
  • @mbaird I think this will end being what needs to be done. Basically using atomic operations in redis and setting a lower TTL on it (which is updated while being processed). Could simply use INCR with a unique key based on the video guid and check if it exists or not. If TTL on this is say 20s and TTL on SQS is 1m, both being updated while a job is being processed every 10s, I think that should solve the issues of dedup and also allow for retries of SQS.
    – dzm
    Dec 1, 2015 at 19:01

5 Answers 5

3

IMO, creating unlimited amount of queues with a single message in each is a really bad design, even if theoretically it would work.

If it was me, I'd try to make sure each video had some sort of unique identifier, that was the same even if the user 'double-clicked' the process button.

I would invision a system where the video, with a unique name (such as a guid) was uploaded to S3, a message gets put in the queue, your threads pickup the message from the queue and do the encoding and then write the video back to a different S3 bucket, but with the same base name.

Before processing any video, I would first check the 'output bucket' to see if there is already an encoded video there, with the matching name, and if it was - I'd skip the reprocessing and delete the message.

If everything is running on an EC2 local disk (and you are not using S3), then the same could be done using an input and output directory on the hard disk (but that would assume that multiple machines aren't doing the processing.

Its important to remember, that its possible for the same message to get delivered by SQS - even if the user only submitted it once. It happens, though rarely, so whatever system you setup you need to make sure if/when you do get the occassional duplicate it doesn't break anything.

1
  • So each video does have a unique guid, unfortunately we're not within AWS so some of the more ideal workflows won't work. But even with a unique guid, checking for the existence of an encoded video, doesn't work since it can take some time while the video is being encoded before it would show up. Ideally, there'd be a mechanism to say "is this video guid in queue or being processed" in an atomic manor. We can definitely use another service or DB, but then it's not as tightly coupled to SQS as I'd like, where there could be false positives, which I've experienced using other queue methods.
    – dzm
    Dec 1, 2015 at 17:52
2

There is no way to ensure the uniqueness of messages in an SQS Queue, or ordering for that matter. Also, having too many queues isn't a good idea.

In my opinion, you need to add another component to your system. A meta data service of some kind would suffice. It could work something like this:

  • When you create an encoding task (before adding it to SQS), you could write it to your meta data service.
  • When a worker receives an encoding task, it would query the meta data service to see if the task has already been completed
  • When a worker completes an encoding task, it would mark the task as completed in the meta data service

If you're uploading the outputs of these encoding jobs to S3, you could effectively use S3 itself as the meta data service. If each video has a unique name/id, you could save the output in S3 with the key of this unique id. Or set it as an S3 meta-data key value (this would make the file a little harder to find as you can't just query the S3 meta data service). Then, When a worker receives an encoding task, it would check if the file already exists on S3, in which case it would delete the message from SQS and skip the task.

If you're not saving the outputs to S3, you'll probably need to employ a database of some kind. Dynamo DB could probably be helpful in terms of speed and cost.

Hope this helps! :)

5
  • So this is definitely something I thought of, but here's a scenario, for a variety of reasons an active encode can fail (server failure, programming exception, etc.). What's nice about SQS is we can use the Visibility Timeout and update it while it's being encoded (for long encodes), however if anything dies which is unable to be handled, this message is now able to be re-processed, as it should. But, if we have another service checking for existence of jobs, this would prevent it from being re-processed.
    – dzm
    Dec 1, 2015 at 17:58
  • Once option could be to use MongoDB with TTL on the documents and update the TTL, just as the Visibility Timeout is being updated. However, if it ever exceeded the VT with SQS, the message received again would be lost, since we'd mark those duplicates as deleted.
    – dzm
    Dec 1, 2015 at 18:01
  • Of course, the viability timeout and update is why SQS is so wonderful for this kind of batch processing usecase. But I don't entirely understand your first comments You wouldn't be preventing anything from being re-processed and you certainly wouldn't need a whole 'other service' to check for the existence of jobs. You'd just check your meta service as soon as you receive a task. And you would only mark a task as completed if it executed successfully. So theres nothing to stop a failed task from being re-processed
    – mickzer
    Dec 1, 2015 at 18:06
  • Well I don't think that would solve the problem then of preventing multiple messages of being processed. If we're only checking if the task has been completed, then it's still possible for multiple tasks to be running at the same time.
    – dzm
    Dec 1, 2015 at 18:13
  • 1
    Oh I understand, you want to prevent multiple workers running the same task at once. Then you'd definitely need to use something like Dynamo. When each worker receives a message, they would update dynamo with the unique identifier to say that this task is already being processed. Then when another worker comes along and receives a message for the same task, it would query dynamo with the unique id and see if it's already being processed. Think of it like a locking mechanism.
    – mickzer
    Dec 1, 2015 at 18:24
2

SQS Has a Deduplication ID Property. Messages sent with the same deduplication ID within a 5-minute window will be successfully recieved, but not actually added to the queue.

You can use this to prevent extra queuing of the same video.

There is some added complexity, even if the message is processed, additional messages with the same deduplication ID won't get queued until the window has elapsed. Likewise, if you send the same ID after the window has elapsed, the message will queue again, which may also be undesired.

However, rather than maintain your own buffer of queued videos, the Deduplication ID should grant you the behaviour you are requesting.

1

Your suggested solution is a bad design, even if it is possible or not. Following is my approach to the problem.

I will use a database (probably DynamoDB) to store an unique id based on the encoded type of the video and I will add a column called status. As soon as user click on the convert button, first, I will check the database. If item is not available, a new record will be pushed to the database with the status "Converting". Then push the work into SQS. After processing the workload, change the status of the database to "Finished". If user click again on the convert button show the result based on the status variable in the database.

7
  • What would happen if the job fails and SQS needs to retry. The status in dynamo would be "Converting" when the new message is received again.
    – dzm
    Dec 1, 2015 at 18:17
  • It's can be handled by the queue worker. Queue worker will get the task from the queue and start converting. If something goes wrong you can just handle the exception. Your work will be still in the queue as you haven't delete it yet. So queue worker will try again the same work until it is success. After success the work, you can delete the message from the queue and update the database. But be careful about the visibility timeout of the queue for not duplicate the work. Dec 1, 2015 at 18:25
  • Well, not all exceptions in this case can be handled. But, I think what can be done is that since each message has a unique message id, no matter how many times you receive it. That message id can be associated with the video guid. When a message is received, it will check if that video guid is there, if it is and the message id does not match, then it's a duplicate... this may work.
    – dzm
    Dec 1, 2015 at 18:30
  • However, since the same message could be received multiple times.. maybe it's not bulletproof.
    – dzm
    Dec 1, 2015 at 18:33
  • I don't have any idea about how your convert process works and why you can't handle the exceptions. What I suggest was the more architectural correct way of doing that. Even though you haven't handle the exceptions, unless you delete the your work from the queue, only one item will be there right? So you can take that advantage without any hazel. Dec 1, 2015 at 18:37
1

There is a way though to check only for unique message after reaceiving data from queue. I will explain the same below.

Let's say you are adding random messages (irrespective of any id or, anything) frequently to a single SQS queue. The logic is for at the time of receiving messages from the queue.

While creating the ReceiveMessageRequest object, you can specify the AttributeNames. So, add "ApproximateReceiveCount" attribute to request object. That will fetch the "ApproximateReceiveCount" value along with each message fetched from the SQS queue.

Now, for the messages which have been read for the first time, the "ApproximateReceiveCount" is 1. Otherwise this value will be greater than 1. So, you can consider only those mesages each time you do a SQS read. Just limit the maximum number of messages read each time by setting the "MaxNumberOfMessages" property of the request object to make sure that you don't get a huge payload on each read (Each 64 KB chunk of a payload is billed as 1 request).

I know, FIFO queue will do a much better job in some cases. But, it has few limitations-

  • It has limited throughput (only 300 transactions per second (TPS))
  • Currently it has support for only two regions (US West (Oregon) and US East (Ohio) regions)

Please find the C# code bellow explaining the logic-

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Amazon.SQS;
using Amazon.SQS.Model;

namespace DriverDataPooler1
{
    class Program
    {
        AmazonSQSClient objClient = new AmazonSQSClient
                ("<AWSAccessKeyId>", "<AWSSecretAccessKey>", Amazon.RegionEndpoint.APSouth1);
        //Create New SQS Queue
        CreateQueueResponse queueResponse = new CreateQueueResponse();
        ListQueuesResponse objqueuesResponseList = new ListQueuesResponse();

        // Declare the request and response objects
        ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest();
        ReceiveMessageResponse receiveMessageResponse = new ReceiveMessageResponse();

        static void Main(string[] args)
        {
            Program p1 = new Program();
            p1.getQueueData();
        }

        public void getQueueData(){

            objqueuesResponseList = objClient.ListQueues(new ListQueuesRequest());
            List<String> QueueList = objqueuesResponseList.QueueUrls;



            // Receive Message from SQS Queue
            if (QueueList.Any())
            {
                // I am only considering the first queue here as I have only one SQS queue
                receiveMessageRequest.QueueUrl = QueueList[0];
                receiveMessageRequest.WaitTimeSeconds = 20;

                //You can limit t6he number of messages to decrease the mayload amount (depends on the size of each message) 
                receiveMessageRequest.MaxNumberOfMessages = 10;
                receiveMessageRequest.AttributeNames = new List<string>() { "ApproximateReceiveCount" };
                receiveMessageResponse = objClient.ReceiveMessage(receiveMessageRequest);
                List<Message> result = receiveMessageResponse.Messages;
                if (result.Any())
                {
                    foreach (Message res in result)
                    {
                        // Checking for the messages that are read for the first time
                        if (Int16.Parse(res.Attributes["ApproximateReceiveCount"]) == 1)

                            // Process you messages here 
                            Console.WriteLine(res.Body);
                    }
                }
                else
                {
                    Console.WriteLine("You have no new messages in your SQS");
                }
            }
            else
            {
                Console.WriteLine("You have no available SQS");
            }
            Console.ReadKey();

        }
    }
}

Please comment if you have any further query.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.