<commit_before>import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.lang.InterruptedException;
import java.net.InetSocketAddress;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.concurrent.ThreadLocalRandom;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
import java.util.TimeZone;
import java.util.TreeMap;
import java.util.stream.Collectors;

import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.*;
import com.datastax.driver.core.policies.*;

import com.google.common.base.Joiner;


class Cass {
	// What Cassandra Ec2Snitch returns
	private static String _localDc = null;

	private static List<String> _remoteDCs = null;
	private static List<String> _allDCs = null;

	// Partial replication
	private static String _ks_pr = null;

	// Attribute popularity keyspace. A table per attribute.
	private static String _ks_attr_pop = null;
	// Object location keyspace.
	private static String _ks_obj_loc  = null;
	private static String _ks_exe_barrier = null;
	private static String _ks_exp_meta = null;

	// For comparison
	private static String _ks_regular = null;

	public static void Init() throws Exception {
		try (Cons.MT _ = new Cons.MT("Cass Init ...")) {
			_WaitUntilYouSeeAllDCs();

			String ks_prefix = "acorn";
			_ks_pr          = ks_prefix + "_pr";
			_ks_attr_pop    = ks_prefix + "_attr_pop";
			_ks_obj_loc     = ks_prefix + "_obj_loc";
			_ks_exe_barrier = ks_prefix + "_exe_barrier";
			_ks_exp_meta    = ks_prefix + "_exp_meta";
			_ks_regular     = ks_prefix + "_regular";
		}
	}

	private static void _WaitUntilYouSeeAllDCs() throws Exception {
		try (Cons.MT _ = new Cons.MT("Wait until you see all DCs ...")) {
			ResultSet rs = _GetSession().execute("select data_center from system.local;");
			// Note that calling rs.all() for the second time returns an empty List<>.
			List<Row> rows = rs.all();
			if (rows.size() != 1)
				throw new RuntimeException(String.format("Unexpcted: %d", rows.size()));

			_localDc = rows.get(0).getString("data_center");
			Cons.P("Local DC: %s", _localDc);

			Cons.Pnnl("Remote DCs:");
			long bt = System.currentTimeMillis();
			boolean first = true;
			while (true) {
				rs = _GetSession().execute("select data_center from system.peers;");
				rows = rs.all();
				if (rows.size() == DC.remoteDCs.size())
					break;

				if (first) {
					System.out.print(" ");
					first = false;
				}
				System.out.print(".");
				System.out.flush();

				if (System.currentTimeMillis() - bt > 2000) {
					System.out.printf("\n");
					throw new RuntimeException("Time out :(");
				}

				Thread.sleep(100);
			}

			_remoteDCs = new ArrayList<String>();
			for (Row r: rows)
				_remoteDCs.add(r.getString("data_center"));
			for (String rDc: _remoteDCs)
				System.out.printf(" %s", rDc);
			System.out.printf("\n");

			_allDCs = new ArrayList<String>();
			_allDCs.add(_localDc);
			for (String rDc: _remoteDCs)
				_allDCs.add(rDc);

			// Prepare remote sessions for later. It takes about 800 ms. Can be
			// threaded to save time, if needed.
			try (Cons.MT _1 = new Cons.MT("Prepareing remote sessions ...")) {
				for (String rDc: _remoteDCs)
					_GetSession(rDc);
			}
		}
	}

	public static String LocalDC() {
		return _localDc;
	}

	public static boolean SchemaExist() throws Exception {
		// Check if the created table, that is created last, exists
		String q = String.format("select * from %s.t0 limit 1", _ks_regular);
		Statement s = new SimpleStatement(q).setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
		try {
			_GetSession().execute(s);
			return true;
		} catch (com.datastax.driver.core.exceptions.InvalidQueryException e) {
			if (e.toString().matches("(.*)Keyspace (.*) does not exist")) {
				return false;
			}
			else if (e.toString().contains("unconfigured table")) {<commit_msg>prepare cassandra sessions in parallel<commit_after>import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.lang.InterruptedException;
import java.net.InetSocketAddress;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.concurrent.ThreadLocalRandom;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
import java.util.TimeZone;
import java.util.TreeMap;
import java.util.stream.Collectors;

import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.*;
import com.datastax.driver.core.policies.*;

import com.google.common.base.Joiner;


class Cass {
	// What Cassandra Ec2Snitch returns
	private static String _localDc = null;

	private static List<String> _remoteDCs = null;
	private static List<String> _allDCs = null;

	// Partial replication
	private static String _ks_pr = null;

	// Attribute popularity keyspace. A table per attribute.
	private static String _ks_attr_pop = null;
	// Object location keyspace.
	private static String _ks_obj_loc  = null;
	private static String _ks_exe_barrier = null;
	private static String _ks_exp_meta = null;

	// For comparison
	private static String _ks_regular = null;

	public static void Init() throws Exception {
		try (Cons.MT _ = new Cons.MT("Cass Init ...")) {
			_WaitUntilYouSeeAllDCs();

			String ks_prefix = "acorn";
			_ks_pr          = ks_prefix + "_pr";
			_ks_attr_pop    = ks_prefix + "_attr_pop";
			_ks_obj_loc     = ks_prefix + "_obj_loc";
			_ks_exe_barrier = ks_prefix + "_exe_barrier";
			_ks_exp_meta    = ks_prefix + "_exp_meta";
			_ks_regular     = ks_prefix + "_regular";
		}
	}

	private static void _WaitUntilYouSeeAllDCs() throws Exception {
		try (Cons.MT _ = new Cons.MT("Wait until you see all DCs ...")) {
			ResultSet rs = _GetSession().execute("select data_center from system.local;");
			// Note that calling rs.all() for the second time returns an empty List<>.
			List<Row> rows = rs.all();
			if (rows.size() != 1)
				throw new RuntimeException(String.format("Unexpcted: %d", rows.size()));

			_localDc = rows.get(0).getString("data_center");
			Cons.P("Local DC: %s", _localDc);

			Cons.Pnnl("Remote DCs:");
			long bt = System.currentTimeMillis();
			boolean first = true;
			while (true) {
				rs = _GetSession().execute("select data_center from system.peers;");
				rows = rs.all();
				if (rows.size() == DC.remoteDCs.size())
					break;

				if (first) {
					System.out.print(" ");
					first = false;
				}
				System.out.print(".");
				System.out.flush();

				if (System.currentTimeMillis() - bt > 2000) {
					System.out.printf("\n");
					throw new RuntimeException("Time out :(");
				}

				Thread.sleep(100);
			}

			_remoteDCs = new ArrayList<String>();
			for (Row r: rows)
				_remoteDCs.add(r.getString("data_center"));
			for (String rDc: _remoteDCs)
				System.out.printf(" %s", rDc);
			System.out.printf("\n");

			_allDCs = new ArrayList<String>();
			_allDCs.add(_localDc);
			for (String rDc: _remoteDCs)
				_allDCs.add(rDc);

			// Prepare remote sessions in parallel for later. It takes about 800 ms
			// per each.
			try (Cons.MT _1 = new Cons.MT("Prepareing remote sessions ...")) {
				List<Thread> threads = new ArrayList<Thread>();
				for (String rDc: _remoteDCs) {
					Thread t = new Thread(new ThreadGetSession(rDc));
					t.start();
					threads.add(t);
				}
				for (Thread t: threads)
					t.join();
			}
		}
	}

	private static class ThreadGetSession implements Runnable
	{
		private String dc;

		ThreadGetSession(String dc) {
			this.dc = dc;
		}

		public void run() {
			try {
				_GetSession(dc);
			} catch (Exception e) {
				System.out.printf("Exception: %s\n%s\n", e, Util.GetStackTrace(e));
				System.exit(1);
			}
		}
	}

	public static String LocalDC() {
		return _localDc;
	}

	public static boolean SchemaExist() throws Exception {
		// Check if the created table, that is created last, exists
		String q = String.format("select * from %s.t0 limit 1", _ks_regular);
		Statement s = new SimpleStatement(q).setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
		try {
			_GetSession().execute(s);
			return true;
		} catch (com.datastax.driver.core.exceptions.InvalidQueryException e) {
			if (e.toString().matches("(.*)Keyspace (.*) does not exist")) {
				return false;
			}
			else if (e.toString().contains("unconfigured table")) {
{'fraction_non_alphanumeric': 0.13028048082427018, 'fraction_numerical': 0.0037779049799656554, 'mean_word_length': 3.6443381180223287, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2549324', 'n_tokens_mistral': 3308, 'n_tokens_neox': 2856, 'n_words': 731}