"Parse Error: Cannot convert to map." when using xread_map

[hmacias-avaya]

I'm trying to understand why an xread_map operation fails with a "Parse Error: Cannot convert to map.".
Looks like I'm not specifying the types correctly somewhere but I can't figure out where.

I'm using something like:

let entries: XReadResponse<String, String, String, RedisValue> = xread_map(None, Some(1000), keys, ids).await?;

where keys and ids are two vectors (with 1 entry if that matters, but there could be many).
This is on a redis cluster with 3 masters and 3 slaves.
I then took the streams example and modified it to do literally the same, just xread_map the exact same vectors.
When I run the streams example, it runs just fine 🤷

If I do the XREAD operations using redis-cli, I get a "CROSSSLOT Keys in request don't hash to the same slot" which I guess is related to the fact I'm using a cluster, but I don't know if this is related to the reason why my code is failing, and if so, how come the streams example does not fail?

does this ring any bells to anybody? any advise on how to troubleshoot this further?

[aembke]

Hey @hmacias-avaya, so the issue is probably related to the CROSSSLOT error you got from redis-cli. That comes from the fact that the keys you're passing to xread_map are located on different hash slots, which the cluster may distribute across different nodes, but that command only goes to one node.

There's a utility function that can make this easier, but you'll end up having to send multiple commands and join the responses together at the end. Something like this is probably what you want:

let keys = vec!["foo", "bar", "baz"]; // etc
let groups = fred::util::group_by_hash_slot(keys)?;

let mut results = Vec::with_capacity(groups.len());
for (_, keys) in groups.into_iter() {
  results.push(client.xread_map(None, Some(1000), keys, ids).await?);
}
// at this point you'd have a `Vec<XReadResponse<...>>`
// then join together the results somehow, depending on whether you'd expect key collisions, etc

In your case it may be slightly more complicated if the IDs are specific to the keys in each group. But ultimately the issue is that each key may be located on a separate cluster node, so the server requires a separate command/query for them. The group_by_hash_slot fn will group the keys by hash slot, but you could also just make a separate call to the server for each key in the worst case.

[hmacias-avaya]

thanks for the fast reply!

I'll give this a try, but then how come the simplified streams example is not showing this same behaviour?

I'll double-check things but I'm literally only doing this:

    let entry: XReadResponse<String, String, String, RedisValue> = client
        .xread_map(None, Some(1000), vec!["SOMETHING"], vec!["SOMETHING_ELSE"])
        .await?;

    for (key, records) in entry.into_iter() {
        for (id, fields) in records.into_iter() {
            println!("Reader recv {} - {}: {:?}", key, id, fields);
        }
    }

(where SOMETHING and SOMETHING_ELSE has been replaced, but I'm using the same exact values in the simplified example and in my bigger code showing the issue)
Both connecting to the same cluster and reading apparently the same 😕

[aembke]

That example assumes a centralized deployment (not clustered), so it avoids this issue. But this kind of thing generally pops up with any command that allows the caller to operate on multiple keys in a single command (MGET, etc). Are you passing in more than one key in your case?

[aembke]

For what it's worth, here's probably how I'd do this. Unfortunately this is probably the most complicated use case for multi-key cluster commands since there's some complexity around managing the mapping between input keys and input IDs.

async fn xread_map_cluster<I: FromValue, K2: FromKey + Hash + Eq, V: FromValue>(
  client: &Client,
  keys: &HashMap<Key, XID>,
  count: Option<u64>,
  block: Option<u64>,
) -> Result<XReadResponse<Key, I, K2, V>, Error> {
  let mut out = HashMap::with_capacity(keys.len());

  // group_by_hash_slot() probably isn't that useful here since XREAD takes both a set of keys and a set of associated
  // IDs, and group_by_hash_slot doesn't preserve the mapping between these inputs. it was intended more for things
  // like MGET, SSUBSCRIBE, etc
  for (key, id) in keys.iter() {
    // only one key will be returned per command so we can use into_iter().next() to read it. but you probably want to
    // do something smarter than unwrap() here if the key is not found.
    let (key, results) = client
      .xread_map::<Key, I, K2, V, _, _>(count, block, key, id)
      .await?
      .into_iter()
      .next()
      .unwrap();

    out.insert(key, results);
  }

  Ok(out)
}

[hmacias-avaya]

thanks again, this looks really helpful.

I forgot to mention the RedisClient creation (I'm on fred 9.4) is the same for my code (failing) and the adapted streams example.
I started with the example (https://github.com/aembke/fred.rs/blob/7a3e97e23fb2467b86ee333b2accebbdb5fddcd2/examples/streams.rs) then trimmed it down to only do the xread part, and to create a RedisClient in the same way. Something like:

let config = RedisConfig::from_url(url)?;
    let client = Builder::from_config(config)
        .set_policy(ReconnectPolicy::default())
        .build()?;

where url is redis-cluster://something.

so, I still do not get how come "redis-cli -c" behaves differently to the modified streams example (also connecting to a cluster) apparently using the same exact arguments.

In my particular case I've started trying to use xread instead of xread_map and I see in some cases what I get back is Null which I'm still investigating. Is it possible that the "Cannot convert to map" is due to this Null and not the fact that I'm using a cluster?

[aembke]

Yeah the nil could also be the issue if the key doesn't exist, or maybe something else is modifying or consuming the key. The cluster thing is a really common issue though so I figured I'd start there.

[hmacias-avaya]

definitely, and again, thanks. I think I have several things to check :)

Just one last question, I see RedisClient derives Clone; how would two copies behave if used concurrently?
can I do xread blocking operations on client1 while I do xadd operations on client2, if they are cloned copies?

In my case I'm currently cloning a client and I'm concurrently using the copies for "an xread loop" while I also xadd using client clones. I'll be testing using clone_new or even a completely new client but I'd like to understand how concurrent uses of a client (or clones of it) would behave.

[aembke]

They share the same connection(s) under the hood, so it depends on a few things. Generally speaking I'd recommend using entirely separate clients/connections if one of them is expected to block though. In your case what will happen is the XADD commands will just wait until the XREAD call finishes or times out, then all the queued XADD commands will run, then it'll do the next XREAD call, etc. In most cases that's not what callers want, so usually I'd recommend using separate connections. The clone_new function can be used to create a new client that will use new connections if you need that - just don't forget to init the new client after calling that.

For what it's worth, this kind of thing is only an issue with blocking commands - sharing shallow clones of clients for other command types is fine, and often recommended since the automatic pipelining will generally save you some network latency. Or alternatively if you're ok with a small lag on the XREADS you could remove the blocking arg and add a sleep between loop iterations. That kind of approach is generally better when the clients are used for other things concurrently, but obviously has some lag trade offs. Otherwise I'd use the two connections approach.

[aembke]

There's also this - https://docs.rs/fred/latest/fred/types/config/enum.Blocking.html. You could set that to Interrupt on the client that does the XREADs, but note that it might affect other use cases too. If you set that then the client will interrupt the XREAD (returning nil) every time you call XADD. That might be what you want too, but it'll result in quite a few more round trips to the server interrupting commands all the time, and ultimately is probably slower than doing the sleep approach.

[hmacias-avaya]

thanks. Indeed, I did a quick test having separate clients and now I do not see what looked like "slow xadd operations" (they were as you said blocked by the blocking xreads and the fact the clients were clones).

Things are starting to make much more sense now, thanks!!