[IGNITE-13213] Improve Queue performance

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 2.4, 2.5, 2.6, 2.7, 2.8, 2.7.5, 2.7.6, 2.8.1
• Component/s: data structures
• Flags: Patch, Important
• Assignee: Unassigned
• Reporter: Manuel Núñez
• Created: 04/Jul/20 13:37
• Updated: 07/Jul/20 09:19

Description

Queue performance can be improved by making some changes, related to Queue Item Key(1) , offer method (2) and collection configuration (3), and related issues IGNITE-13203 and IGNITE-13221.

With those changes queue structure is 10%-30% faster than first Ignite's competitor:

Queue performance benchmark (lower value is better) with different queue message types, multiple clients (M) and 2 server nodes (IMDG):

• string messages: 1KB, 10KB, 100KB
• byte[] messages: 1KB, 10KB, 100KB
• complex objects with cycle references (cpx) messages

Legend:

• Blue: first Ignite's competitor - on heap only, open source version does not support off-heap neither OOTB persistence, we can see GC pauses
• Orange: Ignite off-heap with native persistence enabled
• Red: Ignite off-heap only

1. Queue Item Key -> currently, queue hash code is computed for every single item key using full queue name (> 20 chars) on Queue Item Key constructor.
Proposal compute hash on GridCacheQueueAdapter constructor and pass it to itemKey method

    private static QueueItemKey itemKey(IgniteUuid id,
        int queueNameHash,
        boolean collocated,
        long idx) {
        return collocated ?
            new CollocatedQueueItemKey(id, queueNameHash, idx) : new GridCacheQueueItemKey(id, queueNameHash, idx);
    }

1.1. CollocatedQueueItemKey (fully compatible)

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.datastructures;

import org.apache.ignite.cache.affinity.AffinityKeyMapped;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteUuid;

public class CollocatedQueueItemKey implements QueueItemKey {
    /** */
    private IgniteUuid queueId;

    /** */
    @AffinityKeyMapped
    private int queueNameHash;

    /** */
    private long idx;

    /**
     * @param queueId Queue unique ID.
     * @param queueName Queue name.
     * @param idx Item index.
     */
    public CollocatedQueueItemKey(IgniteUuid queueId, int queueNameHash, long idx) {
        this.queueId = queueId;
        this.queueNameHash = queueNameHash;
        this.idx = idx;
    }

    /** {@inheritDoc} */
    @Override public boolean equals(Object o) {
        if (this == o)
            return true;

        if (o == null || getClass() != o.getClass())
            return false;

        CollocatedQueueItemKey itemKey = (CollocatedQueueItemKey)o;

        return idx == itemKey.idx && queueId.equals(itemKey.queueId);
    }

    /** {@inheritDoc} */
    @Override public int hashCode() {
        int res = queueId.hashCode();

        res = 31 * res + (int)(idx ^ (idx >>> 32));

        return res;
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        return S.toString(CollocatedQueueItemKey.class, this);
    }
}

1.2. GridCacheQueueItemKey (not compatible with existing queues, need to be analysed), remove queueName field as is not required at all (use queueId instead), this will reduce network traffic and improve serialisation performance.

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.datastructures;

import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;

import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;

/**
 * Queue item key.
 */
class GridCacheQueueItemKey implements Externalizable, QueueItemKey {
    /** */
    private static final long serialVersionUID = 0L;

    /** */
    private IgniteUuid queueId;

    /** */
    private long idx;

    /**
     * Required by {@link Externalizable}.
     */
    public GridCacheQueueItemKey() {
        // No-op.
    }

    /**
     * @param queueId Queue unique ID.
     * @param queueNameHash the queue hash, not used
     * @param idx Item index.
     */
    GridCacheQueueItemKey(IgniteUuid queueId, int queueNameHash, long idx) {
        this.queueId = queueId;
        this.idx = idx;
    }

    /**
     * @return Item index.
     */
    public Long index() {
        return idx;
    }

    /**
     * @return Queue UUID.
     */
    public IgniteUuid queueId() {
        return queueId;
    }


    /** {@inheritDoc} */
    @Override public void writeExternal(ObjectOutput out) throws IOException {
        U.writeGridUuid(out, queueId);
        out.writeLong(idx);
    }

    /** {@inheritDoc} */
    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
        queueId = U.readGridUuid(in);
        idx = in.readLong();
    }

    /** {@inheritDoc} */
    @Override public boolean equals(Object o) {
        if (this == o)
            return true;

        if (o == null || getClass() != o.getClass())
            return false;

        GridCacheQueueItemKey itemKey = (GridCacheQueueItemKey)o;

        return idx == itemKey.idx && queueId.equals(itemKey.queueId);
    }

    /** {@inheritDoc} */
    @Override public int hashCode() {
        int res = queueId.hashCode();

        res = 31 * res + (int)(idx ^ (idx >>> 32));

        return res;
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        return S.toString(GridCacheQueueItemKey.class, this);
    }
}

2. Queue's offer method (fully compatible: this change affects GridAtomicCacheQueueImpl and GridTransactionalCacheQueueImpl): Queue structure is implemented as "client side structure", so cache.getAndPut(itemKey(idx), item) call should be replaced by cache.put(itemKey(idx), item) in offer method to avoid unnecessary network traffic.

3. Optional but recommended, currently CollectionConfiguration does not allow to change bounded cache configuration, but on some circustancies could be interesting to improve performance, for example to change CacheWriteSynchronizationMode from default FULL_SYNC to PRIMARY_SYNC, when we have queues with backups.