content: /*
   * Copyright 2012-2025 Aerospike, Inc.
   *
   * Portions may be licensed to Aerospike, Inc. under one or more contributor
   * license agreements WHICH ARE COMPATIBLE WITH THE APACHE LICENSE, VERSION 2.0.
   *
   * Licensed 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 com.aerospike.benchmarks;
  
  
  import com.aerospike.client.AerospikeException;
  import com.aerospike.client.Bin;
  import com.aerospike.client.Key;
  import com.aerospike.client.Record;
  import com.aerospike.client.ResultCode;
  import com.aerospike.client.Value;
  import com.aerospike.client.policy.GenerationPolicy;
  import com.aerospike.client.policy.WritePolicy;
  import com.aerospike.client.util.RandomShift;
  
  /**
   * Random Read/Write workload task.
   */
  public abstract class MRTRWTask {
  
  	final Arguments args;
  	final CounterStore counters;
  	final WritePolicy writePolicyGeneration;
  	final long keyStart;
  	final long keyCount;
  	boolean valid;
  
  	public MRTRWTask(Arguments args, CounterStore counters, long keyStart, long keyCount) {
  		this.args = args;
  		this.counters = counters;
  		this.keyStart = keyStart;
  		this.keyCount = keyCount;
  		this.valid = true;
  		writePolicyGeneration = new WritePolicy(args.writePolicy);
  		writePolicyGeneration.generationPolicy = GenerationPolicy.EXPECT_GEN_EQUAL;
  		writePolicyGeneration.generation = 0;
  	}
  
  	public void stop() {
  		valid = false;
  	}
  
  	protected void runCommand(RandomShift random) throws AerospikeException {
  		try {
  			switch (args.workload) {
  			case READ_UPDATE:
  			case READ_REPLACE:
  				readUpdate(random);
  				break;
  
  			case READ_MODIFY_UPDATE:
  			case READ_MODIFY_INCREMENT:
  			case READ_MODIFY_DECREMENT:
  			case READ_FROM_FILE:
  			case TRANSACTION:
  				throw new Exception("MRT does not support the workload: " + args.workload);
  
  			default:
  				break;
  			}
  		} catch (Exception e) {
  			if (args.debug) {
  				e.printStackTrace();
  			}
  
  		}
  	}
  
  	protected void runNextCommand() {
  	}
  
  	private void readUpdate(RandomShift random) throws AerospikeException {
  		try {
  			if (random.nextInt(100) < args.readPct) {
  				boolean isMultiBin = random.nextInt(100) < args.readMultiBinPct;
  				long key = random.nextLong(keyCount);
  				doRead(key, isMultiBin);
  			} else {
  				boolean isMultiBin = random.nextInt(100) < args.writeMultiBinPct;
  
  				// Perform Single record write even if in batch mode.
  				long key = random.nextLong(keyCount);
  				doWrite(random, key, isMultiBin, null);
  			}
  		} catch (AerospikeException ae) {
  			throw ae;
  		}
  
  	}
  
  
  	protected void doWrite(RandomShift random, long keyIdx, boolean multiBin, WritePolicy writePolicy)
  			throws AerospikeException {
  		Key key = new Key(args.namespace, args.setName, keyStart + keyIdx);
  		// Use predictable value for 0th bin same as key value
  		Bin[] bins = args.getBins(random, multiBin, keyStart + keyIdx);
  
  		try {
  			put(writePolicy, key, bins);
  		} catch (AerospikeException ae) {
  			writeFailure(ae);
  			runNextCommand();
  			throw ae;
  		} catch (Exception e) {
  			writeFailure(e);
  			runNextCommand();
  		}
  	}
  
  	/**
  	 * Read the key at the given index.
  	 */
  	protected void doRead(long keyIdx, boolean multiBin) {
  		try {
  			Key key = new Key(args.namespace, args.setName, keyStart + keyIdx);
  
  			// if udf has been chosen call udf
  			if (args.udfValues != null) {
  				get(key, args.udfPackageName, args.udfFunctionName, args.udfValues);
  			} else if (multiBin) {
  				// Read all bins, maybe validate
  				get(key);
  			} else {
  				// Read one bin, maybe validate
  				get(key, "0");
  			}
  		} catch (AerospikeException ae) {
  			readFailure(ae);
  			runNextCommand();
  		} catch (Exception e) {
  			readFailure(e);
  			runNextCommand();
  		}
  	}
  
  
  	protected void processRead(Key key, Record record) {
  		if (record == null && args.reportNotFound) {
  			counters.readNotFound.getAndIncrement();
  		} else {
  			counters.read.count.getAndIncrement();
  		}
  	}
  
  	protected void processRead(Key key, Object udfReturnValue) {
  		if (udfReturnValue == null && args.reportNotFound) {
  			counters.readNotFound.getAndIncrement();
  		} else {
  			counters.read.count.getAndIncrement();
  		}
  	}
  
  	protected void processBatchRead() {
  		counters.read.count.getAndIncrement();
  	}
  
  	protected void writeFailure(AerospikeException ae) {
  		if (ae.getResultCode() == ResultCode.TIMEOUT) {
  			counters.write.timeouts.getAndIncrement();
  		} else {
  			counters.write.errors.getAndIncrement();
  
  			if (args.debug) {
  				ae.printStackTrace();
  			}
  		}
  	}
  
  	protected void writeFailure(Exception e) {
  		counters.write.errors.getAndIncrement();
  
  		if (args.debug) {
  			e.printStackTrace();
  		}
  	}
  
  	protected void readFailure(AerospikeException ae) {
  		if (ae.getResultCode() == ResultCode.TIMEOUT) {
  			counters.read.timeouts.getAndIncrement();
  		} else {
  			counters.read.errors.getAndIncrement();
  
  			if (args.debug) {
  				ae.printStackTrace();
  			}
  		}
  	}
  
  	protected void readFailure(Exception e) {
  		counters.read.errors.getAndIncrement();
  
  		if (args.debug) {
  			e.printStackTrace();
  		}
  	}
  
  	protected abstract void put(WritePolicy policy, Key key, Bin[] bins);
  
  	protected abstract void add(Key key, Bin[] bins);
  
  	protected abstract void get(Key key, String binName);
  
  	protected abstract void get(Key key);
  
  	protected abstract void get(Key key, String udfPackageName, String udfFunctionName, Value[] udfValues);
  }