2
\$\begingroup\$

I am working with Cassandra where I am getting some stuff out of it using Datastax C# driver.

Below is my code which interacts with Cassandra db and in my GetAsync method I am extracting stuff from three different tables by making multiple async calls for each client id and item id.

Earlier we were using IN clause queries and we found out that performance is not good with that so we change the client code to do multiple async calls in parallel and came up with below code.

public class CassandraUtils
{
    private const string itemMapStmt = "FROM product_data WHERE client_id = ? PER PARTITION LIMIT 1";
    private const string itemSelectStatement = "FROM item WHERE item_id = ? PER PARTITION LIMIT 1";
    private const string itemProStmt = "FROM process_holder WHERE item_id = ? AND process_id = ? PER PARTITION LIMIT 1";

    private readonly IResponseMapper rpm;
    private readonly Cluster cluster;
    private readonly ISession session;
    private readonly IMapper mapper;

    public CassandraUtils(string endpoints, IResponseMapper responseMapper)
    {
        rpm = responseMapper;

        var poolOptions = new PoolingOptions()
            .SetMaxConnectionsPerHost(HostDistance.Remote, 20)
            .SetMaxConnectionsPerHost(HostDistance.Local, 20)
            .SetMaxRequestsPerConnection(1000);

        cluster = Cluster.Builder()
            .AddContactPoints(endpoints)
            .WithApplicationName("Test")
            .WithReconnectionPolicy(new ConstantReconnectionPolicy(100L))
            .WithLoadBalancingPolicy(new TokenAwarePolicy(new DCAwareRoundRobinPolicy("dc5")))
            .WithPoolingOptions(poolOptions)
            .WithQueryOptions(new QueryOptions()
                .SetConsistencyLevel(ConsistencyLevel.LocalQuorum)
                .SetSerialConsistencyLevel(ConsistencyLevel.LocalSerial))
            .WithCompression(CompressionType.LZ4)
            .WithSocketOptions(new SocketOptions().SetReadTimeoutMillis(6000))
            .Build();

        var keyspace = Environment.GetEnvironmentVariable("keyspace_name");
        session = cluster.Connect(keyspace);
        session.UserDefinedTypes.Define(
            UdtMap.For<ASHelper>("as")
                .Map(a => a.Url, "url").Map(a => a.Id, "id"),
            UdtMap.For<MonthlyProcess>("monthly_process")
        );

        mapper = new Mapper(session);
    }

    /**
     *
     * Below method gets data from all 3 different tables by making multiple async calls.
     *
     */
    public async Task<IList<Item>> GetAsync(IList<int> clientIds, int processId, int proc, Kyte kt)
    {
        var clientMaps = await ProcessCassQueries(clientIds, (ct, batch) => mapper.SingleOrDefaultAsync<ItemMapPoco>(itemMapStmt, batch), "GetPIMValue");

        if (clientMaps == null || clientMaps.Count <= 0)
        {
            return null;
        }

        var itemIds = clientMaps.SelectMany(x => x.ItemIds).Where(y => y != null).ToList();

        var itemsTask = ProcessCassQueries(itemIds, (ct, batch) => mapper.SingleOrDefaultAsync<ItemPoco>(itemSelectStmt, batch), "GetAsync");

        var itemProTask = ProcessCassQueries(itemIds, (ct, batch) => mapper.SingleOrDefaultAsync<ItemProPoco>(itemProStmt, batch, processId), "GetIPValue");

        var items = await itemsTask;
        if (items.Count <= 0)
        {
            return null;
        }

        var itmDictionary = items.ToDictionary(dto => dto.ItemId, dto => rpm.MapToItem(dto, proc));
        var itmProDict = itemIds.ToDictionary<int, int, ItemProPoco>(id => id, id => null);
        var holder = new List<int>();

        var itemPrices = await itemProTask;
        itemPrices.ForEach(i => { if (i != null) itmProDict[i.ItemId] = i; });
        foreach (var ip in itmProDict) if (ip.Value == null) holder.Add(ip.Key);

        if (holder.Count > 0)
        {
            var ipHolder = await ProcessCassQueries(itemIds, (ct, batch) => mapper.SingleOrDefaultAsync<ItemProPoco>(itemProStmt, batch, kt.Pid), "GetIPValue");
            ipHolder.ToList().ForEach(ipf => { if (ipf != null) itmProDict[ipf.ItemId] = ipf; });
        }

        return itmDictionary.Select(kvp =>
        {
            itmProDict.TryGetValue(kvp.Key, out var ip);

            if (kvp.Value != null)
            {
                rpm.convert(ip, kvp.Value);
                return kvp.Value;
            }

            return null;
        }).Where(s => s != null).ToList();
    }

    /**
     *
     * Below method does multiple async calls on each table for their corresponding id's.
     *
     */
    private async Task<List<T>> ProcessCassQueries<T>(IList<int> ids, Func<CancellationToken, int, Task<T>> mapperFunc, string msg) where T : class
    {
        var requestTasks = ids.Select(id => ProcessCassQuery(ct => mapperFunc(ct, id), msg));
        return (await Task.WhenAll(requestTasks)).Where(e => e != null).ToList();
    }

    // this might not be good idea to do it
    private Task<T> ProcessCassQuery<T>(Func<CancellationToken, Task<T>> requestExecuter, string msg) where T : class
    {
        return requestExecuter(CancellationToken.None);
    }
}

I am using latest datastax c# driver to interact with cassandra.

Cluster Info:

  • We have a 6 nodes cluster all in one dc with RF as 3.
  • We read/write as local quorum.
  • And we have row caching enabled as well.

Few questions:

I am looking to see if there is anything that can be improved or written efficiently in my GetAsync, ProcessCassQueries and ProcessCassQuery method? Since I am pretty new to C# so it might not be efficient way to do what I am doing currently. I need to execute multiple clientIds and itemIds queries in parallel to cassandra so I came up with above which does the job but not sure whether it is the efficient way to do it.

One thing I realized is - I am doing lot of concurrent async calls for each table to cassandra db which might not be good idea. So maybe I should try to limit the number of async calls I can do using Semaphore? Also I am using lot of expensive LINQ calls like ToDictionary, ToList and ForEach that could be problematic under heavy load?

\$\endgroup\$

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.