Personally, I just think it got expensive for the unbundled mode (now 20$/1m seconds instead of 12.5

Personally, I just think it got expensive for the unbundled mode (now 20$/1m seconds instead of 12.5$ and 0.3$ instead of 0.15$ per 1m requests). I know in the document we talk about the cpu time, instead of clock time, but to be honest my CPU time is usually very low, and thus the only impacting factor for me would be the number of requests, which unfortunately just doubled in price.

That would not change much for me, because Cloudflare is still a very generous platform and is pretty easy to use, I just don't like how this is promoted as only beneficial

Oomar4289 Personally, I just think it got expensive for the unbundled mode (now 20$/1m sec...

Unsmart•10/26/23, 10:17 PM

~100ms of wall time = + $0.15/million request

Also one unknown part of the unbound billing model is if you used 1 second of wall time and 1 second of CPU time you actually get billed 8 seconds wall time instead. (8x multiplier on CPU time)

So unless you had extremely fast requests it will probably be cheaper or around the same price.

UUnsmart ~100ms of wall time = + $0.15/million request Also one unknown part of the unbo...

omar4289OP•10/26/23, 10:21 PM

Lets see, but anyway, I still like the products Cloudflare is working ln. Really good job and I am open to pay any extra amount (if any) for the great services you provide at a very reasonable price

omar4289OP•10/26/23, 10:23 PM

Just one more question, since workers run on a 128MB machines, does it mean that the number of ms that we have is per 1GB of RAM, or is it per worker (128MB)?

Unsmart•10/26/23, 10:25 PM

The unbound billing model is based on 128MB (or 1/8th a GB) so 8 seconds of true wall time = 1 GB-sec for billing

omar4289OP•10/26/23, 10:30 PM

Ok, but right now, assuming I only run a loop in my program without I/O (very unlikely) and that takes exactly 1second to complete.

In the old model, I would pay 12.5$ for 8 million executions
In the new model, I would pay 20$ for 1 million executions.

I know the I/O work usually takes time, but it all depends on what you do, sometimes it will be cheaper, sometimes more expensive. Lets see how it goes

Unsmart•10/26/23, 10:37 PM

You can use this calculator and input your metrics to see the cost difference: https://calculator.imcf.me/

Unsmart•10/26/23, 10:38 PM

(The calculator doesnt take into account the CPU multiplier so you would need to make sure the wall time ms is at least 8x the cpu time ms)

Mad•10/28/23, 11:58 PM

I have a single queue marked both as a producer and a consumer queue, attached to the same worker. The worker takes input from the queue and produces output to the same queue until the task is done. What I observe is, the task runs for couple of iterations and stops running.

Mad•10/28/23, 11:58 PM

works fine when I run this locally

Mad•10/28/23, 11:59 PM

@Charlie B | KV, Queues, PubSub any idea what must be going wrong. Can I not use the same queue as a producer as well as consumer?

paper___•11/2/23, 2:53 AM

Hello, I'm having a problem regarding queues. I have a worker that streams and parses a big CSV file and puts one JSON object per row in a batch of 100 to a Queue. Locally, on dev, the flow works flawlessly, I can parse and enqueue a 250Mb CSV file with no problem and using low heap space. But in cloudflare, for some unknown reason the worker stops enqueuing at around 2600 or so objects with no error or warning whatsoever.

The code of the worker looks like this:

export async fetch(request: Request, env: Environment, ctx: ExecutionContext) {
  const response = await fetch(someUrl);
  const objectsStream = response.body!.getReader()
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(new CSVParserStream());
  const enqueue = async () => {
    for await (const objects of take(100, objectsStream)) {
      await env.MY_QUEUE.sendBatch(objects.map(object => ({
        body: object,
        contentType: "json",
      })));
    }
  };
  ctx.waitUntil(
    enqueue()
      .then(() => {
        console.log("Success");
      })
      .catch(console.error)
  );
  return new Response("OK");
}

export async fetch(request: Request, env: Environment, ctx: ExecutionContext) {
  const response = await fetch(someUrl);
  const objectsStream = response.body!.getReader()
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(new CSVParserStream());
  const enqueue = async () => {
    for await (const objects of take(100, objectsStream)) {
      await env.MY_QUEUE.sendBatch(objects.map(object => ({
        body: object,
        contentType: "json",
      })));
    }
  };
  ctx.waitUntil(
    enqueue()
      .then(() => {
        console.log("Success");
      })
      .catch(console.error)
  );
  return new Response("OK");
}

kian•11/2/23, 5:55 AM

Is that 2600 objects taking 30 seconds by any chance?

Kkian Is that 2600 objects taking 30 seconds by any chance?

paper___•11/2/23, 2:24 PM

Not really, it is 5 seconds top

Ppaper___Hello, I'm having a problem regarding queues. I have a worker that streams and p...

robmarscher•11/3/23, 4:31 PM

I had a similar issue that I think was due to the 1000 max subrequests per worker invocation. I think the queue send counts as a subrequest. https://developers.cloudflare.com/workers/platform/limits/#how-many-subrequests-can-i-make

That doesn't add up with 2600 objects though... that would only be 26 requests with a batch size of 100. 128 MB is the memory limit... but I assume that wouldn't be hit when you are streaming the body.

With sendBatch, there is a 256KB max size... maybe you are hitting that at a certain point in your data set and reducing the number of objects in sendBatch would fix?

A batch can contain up to 100 messages, though items are limited to 128 KB each, and the total size of the array cannot exceed 256 KB.

https://developers.cloudflare.com/queues/platform/javascript-apis/#queue

For the subrequest limit, I solved it by storing the uploaded csv on R2 and then using another queue to read the csv. That queue handler would track how many rows it processed and then once it reached 750 (arbitrary number to leave buffer under the 1000 limit) it would enqueue another task to read the csv - passing the number of rows to skip and the path to the csv on R2 in the queue message data. I also used a Durable Object to track the progress and push updates via websocket to the client that uploaded the csv.

Ppaper___Hello, I'm having a problem regarding queues. I have a worker that streams and p...

Unsmart•11/3/23, 4:35 PM

Try running the code without using ctx.waitUntil and see if you get any errors... its possible that waitUntil is swallowing the error

UUnsmart Try running the code without using ctx.waitUntil and see if you get any errors.....

Matt Silverlock•11/3/23, 7:54 PM

This is very common - you will swalllow the exception inside

waitUntil

waitUntil

unless you are actively logging

cg•11/3/23, 9:02 PM

Are Queues down at the moment?

Ccg Are Queues down at the moment?

Matt Silverlock•11/4/23, 11:36 AM

We observed a small blip affecting some Queues around the time of your post.

Rrobmarscher I had a similar issue that I think was due to the 1000 max subrequests per worke...

paper___•11/6/23, 12:42 PM

Thanks Rob! I'm going to try checking the size of the batch and see if I'm hitting the limit. When you say 128kb each message, that is JSON encoded right?

UUnsmart Try running the code without using ctx.waitUntil and see if you get any errors.....

paper___•11/6/23, 1:19 PM

 [ERROR]   Error: Worker exceeded memory limit.

 [ERROR]   Error: Worker exceeded memory limit.

paper___•11/6/23, 1:20 PM

The only thing I'm thinking might be the problem is a mem leak in the csv parsing function

Ppaper___The only thing I'm thinking might be the problem is a mem leak in the csv parsin...

Matt Silverlock•11/6/23, 1:36 PM

If your decoder isn’t streaming correctly you will definitely run out of memory.

MMatt Silverlock If your decoder isn’t streaming correctly you will definitely run out of memory.

paper___•11/6/23, 3:26 PM

I'm using

csv-parse

csv-parse

, maybe the internal Buffer is not flushing correctly.

paper___•11/6/23, 3:26 PM

I'm creating a snapshot right now to check that

Nnil Hello everyone 🙂 I am having an issue with concurrency, for some reason my que...

Matt Silverlock•11/7/23, 5:51 PM

What is your account ID, and what are the settings for
max_batch_size
max_batch_size
and
max_batch_timeout
max_batch_timeout
on the queue? (I see them now, missed those).

What is your message volume during that period? 567 messages in a 24 hour period isn't a lot, and so your consumer shouldn't have to scale up. If your consumer is making calls to an external service before ack'ing that will also increase the measured delay.

Nnil Thanks Matt! This is my account ID: `34c0d8412965b33f782ad043d255542c` the queue...

Matt Silverlock•11/7/23, 5:58 PM

Can you share your message rate chart as well, for the same time period?

Nnil also regarding what you said, indeed before ack'ing the messages multiple calls ...

Matt Silverlock•11/7/23, 6:06 PM

No: batch timeout is on the producer side. It does not determine how the consumer scales. Your volume/message rate is very low, and (right now) we prefer to avoid scaling too quickly on the consumer side.

jasonkuhrt•11/7/23, 6:29 PM

Question about queues I guess I should have put here? workers-and-pages-helpAre queue producers and consumers supposed to work in local development with wrangler?

AKStriker98•11/8/23, 5:18 PM

Question about the queues beta. Given the public beta has been out for almost a year now, is there a rough timeline or progress level for how far along the public-beta is for full production? We would love to use queues for logging tasks but don’t yet feel comfortable using a beta product.

AAKStriker98 Question about the queues beta. Given the public beta has been out for almost a ...

Matt Silverlock•11/8/23, 8:19 PM

We want to get Queues into GA by end of Q1 next year. There’s architectural work to get per-queue throughput into the thousands-per-queue that’s happening behind the scenes.

lenqbit•11/9/23, 9:12 AM

Hello, what is the quickest/fastest way to purge a queue? We are in a situation where our queue has accumulated too much backlog that our consumer can not cope with it anymore.
We would like to purge the whole queue first as way to mitigate the issue before we look at finding out the underlying issue and work on a long term fix.
Is it a case of deploying a new version of our consumer that just acks the batches with say

ackAll()

ackAll()

?
Or is there an API that can be invoked from outside a worker?

lenqbit•11/9/23, 9:13 AM

I've seen that there is a way to list and ack message in the dashboard but this will not help as it limits the batch size to 100

Llenqbit Hello, what is the quickest/fastest way to purge a queue? We are in a situation ...

Matt Silverlock•11/9/23, 11:20 AM

There’s not a “purge all” method - re-deploying a consumer that acks every batch and returns immediately is the fastest way right now. Or deleting the queue/rebinding to a new queue.

rikka0w0•11/9/23, 1:01 PM

According to the JavaScript API documentation of Queue, inside the 'export default' block, the queue consumer is a separate handler(queue). Is there a way to poll a queue inside a running fetch()? Or it is possible to get notified inside a running fetch function for changes of a queue? Im looking for a way to implement communication between two running worker instances.

kian•11/9/23, 1:02 PM

Not without using something like Durable Objects where the Workers can read/write state for the other

Kkian Not without using something like Durable Objects where the Workers can read/writ...

rikka0w0•11/9/23, 2:32 PM

Will it be possible to implement such function with kv?

kian•11/9/23, 2:43 PM

It’s not ideal because KV is eventually consistent and reads are cached for up to 60 seconds, so polling doesn’t really work

kian•11/9/23, 2:44 PM

If you polled, and there was nothing, it’d still read as nothing even if your queue wrote to KV for up to 60 seconds.

Stockton•11/9/23, 4:55 PM

Are Queues right for me?

I'm looking to send many events to a queue, wait 30 seconds, and then send a single array with all accumulated events to a destination.

When I removed the batch.messages for each part so that it wouldn't process each event individually, it failed.

export async fetch(request: Request, env: Environment, ctx: ExecutionContext) { const response = await fetch(someUrl); const objectsStream = response.body!.getReader() .pipeThrough(new TextDecoderStream()) .pipeThrough(new CSVParserStream()); const enqueue = async () => { for await (const objects of take(100, objectsStream)) { await env.MY_QUEUE.sendBatch(objects.map(object => ({ body: object, contentType: "json", }))); } }; ctx.waitUntil( enqueue() .then(() => { console.log("Success"); }) .catch(console.error) ); return new Response("OK"); }

Personally, I just think it got expensive for the unbundled mode (now 20$/1m seconds instead of 12.5

Similar Threads

Similar Threads

Similar Threads