#include <bits/stdc++.h>
#include "algotester.h"
using namespace std;
namespace {
constexpr int MAX_ID = 2147483647;
constexpr int T_ROUNDS = 86400;
struct Query {
int id;
long long cpu;
long long mem;
long long qto;
int type; // 0 write, 1 read
};
struct ServerModel {
int id;
long long cpu;
long long mem;
long long cost;
};
struct Instance {
int id;
int level;
int b;
vector<int> initialNodes;
int primaryNodeId;
int activeNodes = 0;
};
struct RequestGroup {
int instanceId;
int queryId;
long long qps;
};
struct MiddlewareChunk {
long long arrivalTime;
long long count;
};
struct ServerState {
int id;
int modelId;
long long capCpu;
long long capMem;
long long costPerSecond = 0;
long long usedCpu = 0;
long long usedMem = 0;
int activeNodeCount = 0;
int reservationCount = 0;
bool exists = false;
};
struct QueueChunk {
int instanceId;
int queryId;
long long arrivalTime;
long long count;
};
struct NodeState {
int id = -1;
int instanceId = -1;
bool primary = false;
bool initialized = false; // placed initially or created via scale out
bool active = false; // can receive dispatch
int serverId = -1;
long long cpu = 0;
long long mem = 0;
bool pending = false;
int pendingEffectTime = -1;
deque<QueueChunk> queue;
long long headRemainingCpu = 0;
};
enum class OpType {
LOCAL,
CROSS,
SCALE_OUT,
};
struct PendingEvent {
OpType type;
int nodeId;
int effectTime;
// Common: target node final spec (for LOCAL/CROSS) or created spec (SCALE_OUT)
long long newCpu;
long long newMem;
// CROSS/SCALE_OUT
int targetServerId = -1;
bool hasReservation = false;
// SCALE_OUT
int instanceId = -1;
};
struct ParsedInput {
vector<ServerModel> serverModels;
vector<Query> queries;
vector<Instance> instances;
vector<vector<RequestGroup>> arrivalsByRound; // 1..T_ROUNDS
long long cpuMin = 0;
long long memMin = 0;
long long cpuMax = 0;
long long memMax = 0;
long long gCpu = 0;
long long gMem = 0;
int Tlu = 0;
int Tcu = 0;
int Tso = 0;
array<int, 9> weights{};
};
long long pairKey(int a, int b) {
return (static_cast<long long>(a) << 32) ^ static_cast<unsigned int>(b);
}
long long nonNegativeInt(AlgotesterReader& out, const string& name) {
return out.readInt(0, LLONG_MAX / 4, name);
}
bool isMultiple(long long x, long long g) {
return g != 0 && x % g == 0;
}
} // namespace
int main(int argc, char* argv[]) {
ios::sync_with_stdio(false);
auto [test_in, user_in, user_out] = initInteractiveChecker(argc, argv);
auto score_log = ofstream(argc >= 6 ? argv[5] : "/dev/null");
auto error = [&user_in]() {
user_in << -1 << '\n';
user_in.flush();
};
user_out.setErrorCallback(error);
auto ensure = [&error](bool cond, const string& msg) {
check(cond, msg, error);
};
ParsedInput in;
in.arrivalsByRound.assign(T_ROUNDS + 1, {});
// --- Parse generator/validator format from test input.
int n = static_cast<int>(test_in.readInt(1, 5, "n_servers"));
test_in.readEndl();
in.serverModels.reserve(n);
unordered_map<int, int> modelIndex;
for (int i = 0; i < n; ++i) {
ServerModel s;
s.id = static_cast<int>(test_in.readInt(1, MAX_ID, "server_id"));
test_in.readSpace();
s.cpu = test_in.readInt(320000, 1180000, "server_cpu");
test_in.readSpace();
s.mem = test_in.readInt(64000, 1024000, "server_mem");
test_in.readSpace();
s.cost = test_in.readInt(10, 1000, "server_cost");
test_in.readEndl();
ensure(!modelIndex.count(s.id), "duplicate server model id");
modelIndex[s.id] = static_cast<int>(in.serverModels.size());
in.serverModels.push_back(s);
}
int k = static_cast<int>(test_in.readInt(1, 100, "k_queries"));
test_in.readEndl();
in.queries.reserve(k);
unordered_map<int, int> queryIndex;
for (int i = 0; i < k; ++i) {
Query q;
q.id = static_cast<int>(test_in.readInt(1, MAX_ID, "query_id"));
test_in.readSpace();
q.cpu = test_in.readInt(10, 100000000, "query_cpu");
test_in.readSpace();
q.mem = test_in.readInt(1, 100000, "query_mem");
test_in.readSpace();
q.qto = test_in.readInt(0, 3600, "query_qto");
test_in.readSpace();
q.type = static_cast<int>(test_in.readInt(0, 1, "query_type"));
test_in.readEndl();
ensure(!queryIndex.count(q.id), "duplicate query id");
queryIndex[q.id] = static_cast<int>(in.queries.size());
in.queries.push_back(q);
}
int m = static_cast<int>(test_in.readInt(1, 10, "m_instances"));
test_in.readEndl();
in.instances.reserve(m);
unordered_map<int, int> instanceIndex;
unordered_map<int, NodeState> nodes;
nodes.reserve(200000);
for (int i = 0; i < m; ++i) {
Instance inst;
inst.id = static_cast<int>(test_in.readInt(1, MAX_ID, "instance_id"));
test_in.readSpace();
inst.level = static_cast<int>(test_in.readInt(1, 9, "level"));
test_in.readSpace();
inst.b = static_cast<int>(test_in.readInt(1, 100000, "initial_nodes_count"));
ensure(!instanceIndex.count(inst.id), "duplicate instance id");
instanceIndex[inst.id] = static_cast<int>(in.instances.size());
inst.initialNodes.reserve(inst.b);
for (int j = 0; j < inst.b; ++j) {
test_in.readSpace();
int nodeId = static_cast<int>(test_in.readInt(1, MAX_ID, "node_id"));
ensure(!nodes.count(nodeId), "duplicate initial node id");
NodeState nd;
nd.id = nodeId;
nd.instanceId = inst.id;
nd.primary = (j == 0);
nodes[nodeId] = nd;
inst.initialNodes.push_back(nodeId);
if (j == 0) inst.primaryNodeId = nodeId;
}
test_in.readEndl();
in.instances.push_back(inst);
}
int r = static_cast<int>(test_in.readInt(0, T_ROUNDS, "request_timestamps_count"));
test_in.readEndl();
for (int i = 0; i < r; ++i) {
string line = test_in.readLine();
ensure(!line.empty(), "empty request line");
stringstream ss(line);
vector<long long> vals;
long long x;
while (ss >> x) vals.push_back(x);
ensure(!vals.empty(), "invalid request line");
ensure(vals.size() >= 4, "request line must contain at least one group");
ensure((vals.size() - 1) % 3 == 0, "invalid request line arity");
int ts = static_cast<int>(vals[0]);
ensure(0 <= ts && ts < T_ROUNDS, "request timestamp out of range");
int round = ts + 1;
for (size_t p = 1; p < vals.size(); p += 3) {
int instId = static_cast<int>(vals[p]);
int qId = static_cast<int>(vals[p + 1]);
long long qps = vals[p + 2];
ensure(instanceIndex.count(instId), "request references unknown instance");
ensure(queryIndex.count(qId), "request references unknown query");
ensure(1 <= qps && qps <= 10000, "request qps out of range");
in.arrivalsByRound[round].push_back({instId, qId, qps});
}
}
in.cpuMin = test_in.readInt(100, 1000000000, "cpu_min");
test_in.readSpace();
in.memMin = test_in.readInt(100, 1000000000, "mem_min");
test_in.readEndl();
in.cpuMax = test_in.readInt(100, 1000000000, "cpu_max");
test_in.readSpace();
in.memMax = test_in.readInt(100, 1000000000, "mem_max");
test_in.readEndl();
ensure(in.cpuMin <= in.cpuMax, "cpu_min must be <= cpu_max");
ensure(in.memMin <= in.memMax, "mem_min must be <= mem_max");
in.gCpu = test_in.readInt(100, 5000, "g_cpu");
test_in.readSpace();
in.gMem = test_in.readInt(100, 5000, "g_mem");
test_in.readEndl();
in.Tlu = static_cast<int>(test_in.readInt(1, 100, "T_lu"));
test_in.readEndl();
in.Tcu = static_cast<int>(test_in.readInt(1, 100, "T_cu"));
test_in.readEndl();
in.Tso = static_cast<int>(test_in.readInt(1, 100, "T_so"));
test_in.readEndl();
int prevW = 0;
for (int i = 0; i < 9; ++i) {
in.weights[i] = static_cast<int>(test_in.readInt(1, 100, "L_i"));
ensure(in.weights[i] >= prevW, "weights must be non-decreasing");
prevW = in.weights[i];
if (i + 1 < 9) test_in.readSpace();
else test_in.readEndl();
}
test_in.readEof();
// --- Send static input to participant in statement protocol order.
user_in << in.queries.size() << '\n';
for (const auto& q : in.queries) {
user_in << q.id << ' ' << q.cpu << ' ' << q.mem << ' ' << q.qto << ' ' << q.type << '\n';
}
user_in << in.instances.size() << '\n';
for (const auto& inst : in.instances) {
user_in << inst.id << ' ' << inst.level << ' ' << inst.b;
for (int nodeId : inst.initialNodes) user_in << ' ' << nodeId;
user_in << '\n';
}
user_in << in.serverModels.size() << '\n';
for (const auto& s : in.serverModels) {
user_in << s.id << ' ' << s.cpu << ' ' << s.mem << ' ' << s.cost << '\n';
}
user_in << in.cpuMin << ' ' << in.memMin << '\n';
user_in << in.cpuMax << ' ' << in.memMax << '\n';
user_in << in.gCpu << ' ' << in.gMem << '\n';
user_in << in.Tlu << ' ' << in.Tcu << ' ' << in.Tso << '\n';
for (int i = 0; i < 9; ++i) {
if (i) user_in << ' ';
user_in << in.weights[i];
}
user_in << '\n';
user_in << T_ROUNDS << '\n';
user_in.flush();
// --- Runtime state.
unordered_map<int, ServerState> servers;
servers.reserve(200000);
long double runningCostPerRound = 0.0L;
long double C_r = 0.0L;
long double C_s = 0.0L;
auto getModel = [&](int modelId) -> const ServerModel& {
ensure(modelIndex.count(modelId), "unknown server model id");
return in.serverModels[modelIndex[modelId]];
};
auto ensureServerExists = [&](int serverId) -> ServerState& {
ensure(servers.count(serverId) && servers[serverId].exists, "server does not exist");
return servers[serverId];
};
auto ensureNodeExists = [&](int nodeId) -> NodeState& {
ensure(nodes.count(nodeId), "unknown node id");
return nodes[nodeId];
};
// Only ids chosen by participant: NEW_SERVER and SCALE_OUT node ids.
unordered_set<int> chosenIds;
chosenIds.reserve(200000);
// Middleware buffer per (instance, query).
unordered_map<long long, deque<MiddlewareChunk>> middleware;
middleware.reserve(1 << 14);
// Pending events by effect time.
int maxDelay = max({in.Tlu, in.Tcu, in.Tso});
vector<vector<PendingEvent>> events(T_ROUNDS + maxDelay + 5);
auto addMiddleware = [&](int instanceId, int queryId, long long arrivalTime, long long amount) {
long long key = pairKey(instanceId, queryId);
auto& dq = middleware[key];
if (!dq.empty() && dq.back().arrivalTime == arrivalTime) {
dq.back().count += amount;
} else {
dq.push_back({arrivalTime, amount});
}
};
auto consumeMiddleware = [&](int instanceId, int queryId, long long amount) -> vector<MiddlewareChunk> {
long long key = pairKey(instanceId, queryId);
ensure(middleware.count(key), "dispatch uses empty middleware buffer");
auto& dq = middleware[key];
long long rem = amount;
vector<MiddlewareChunk> taken;
while (rem > 0 && !dq.empty()) {
long long take = min(rem, dq.front().count);
taken.push_back({dq.front().arrivalTime, take});
dq.front().count -= take;
rem -= take;
if (dq.front().count == 0) dq.pop_front();
}
ensure(rem == 0, "dispatch amount exceeds middleware buffer for (instance, query)");
if (dq.empty()) middleware.erase(key);
return taken;
};
auto enqueueToNode = [&](NodeState& nd, int instanceId, int queryId, long long arrivalTime, long long amount) {
ensure(amount > 0, "dispatch amount must be positive");
if (!nd.queue.empty() &&
nd.queue.back().instanceId == instanceId &&
nd.queue.back().queryId == queryId &&
nd.queue.back().arrivalTime == arrivalTime) {
nd.queue.back().count += amount;
} else {
nd.queue.push_back({instanceId, queryId, arrivalTime, amount});
}
};
auto serverCanFit = [&](const ServerState& s, long long addCpu, long long addMem) {
if (addCpu < 0 || addMem < 0) return false;
return s.usedCpu + addCpu <= s.capCpu && s.usedMem + addMem <= s.capMem;
};
auto validateFinalSpec = [&](long long cpu, long long mem, bool allowZero) {
if ((cpu == 0) != (mem == 0)) return false;
if (cpu == 0 && mem == 0) return allowZero;
if (cpu < in.cpuMin || cpu > in.cpuMax) return false;
if (mem < in.memMin || mem > in.memMax) return false;
return true;
};
auto applyEventAtRound = [&](const PendingEvent& ev) {
NodeState& nd = ensureNodeExists(ev.nodeId);
ensure(nd.pending && nd.pendingEffectTime == ev.effectTime, "inconsistent pending node state");
int instIdx = instanceIndex[nd.instanceId];
Instance& inst = in.instances[instIdx];
if (ev.type == OpType::LOCAL) {
ensure(nd.active, "local scale on inactive node at effect time");
ensure(nd.serverId != -1, "local scale node has no server at effect time");
ServerState& srv = ensureServerExists(nd.serverId);
long long oldCpu = nd.cpu;
long long oldMem = nd.mem;
if (ev.newCpu == 0) {
ensure(!nd.primary, "primary node cannot be released");
ensure(inst.activeNodes - 1 >= inst.b, "instance falls below initial node count at effect time");
ensure(nd.queue.empty() && nd.headRemainingCpu == 0,
"cannot release node with unfinished requests at effect time");
srv.usedCpu -= oldCpu;
srv.usedMem -= oldMem;
srv.activeNodeCount--;
nd.active = false;
nd.cpu = 0;
nd.mem = 0;
nd.serverId = -1;
inst.activeNodes--;
} else {
long long nextUsedCpu = srv.usedCpu - oldCpu + ev.newCpu;
long long nextUsedMem = srv.usedMem - oldMem + ev.newMem;
ensure(nextUsedCpu <= srv.capCpu && nextUsedMem <= srv.capMem,
"local scale causes server over-allocation at effect time");
srv.usedCpu = nextUsedCpu;
srv.usedMem = nextUsedMem;
nd.cpu = ev.newCpu;
nd.mem = ev.newMem;
}
nd.pending = false;
nd.pendingEffectTime = -1;
return;
}
if (ev.type == OpType::CROSS) {
ensure(nd.active, "cross scale on inactive node at effect time");
ensure(nd.serverId != -1, "cross scale node has no source server at effect time");
ServerState& src = ensureServerExists(nd.serverId);
long long oldCpu = nd.cpu;
long long oldMem = nd.mem;
if (ev.newCpu == 0) {
ensure(!nd.primary, "primary node cannot be released");
ensure(inst.activeNodes - 1 >= inst.b, "instance falls below initial node count at effect time");
ensure(nd.queue.empty() && nd.headRemainingCpu == 0,
"cannot release node with unfinished requests at effect time");
src.usedCpu -= oldCpu;
src.usedMem -= oldMem;
src.activeNodeCount--;
nd.active = false;
nd.cpu = 0;
nd.mem = 0;
nd.serverId = -1;
inst.activeNodes--;
nd.pending = false;
nd.pendingEffectTime = -1;
return;
}
ServerState& tgt = ensureServerExists(ev.targetServerId);
ensure(ev.hasReservation, "cross effect missing reservation");
src.usedCpu -= oldCpu;
src.usedMem -= oldMem;
src.activeNodeCount--;
tgt.reservationCount--;
tgt.activeNodeCount++;
nd.serverId = ev.targetServerId;
nd.cpu = ev.newCpu;
nd.mem = ev.newMem;
nd.pending = false;
nd.pendingEffectTime = -1;
return;
}
// SCALE_OUT
ensure(ev.type == OpType::SCALE_OUT, "unknown event type");
ensure(!nd.active, "scale-out node unexpectedly active before effect");
ensure(nd.serverId == ev.targetServerId, "scale-out node server mismatch");
ServerState& tgt = ensureServerExists(ev.targetServerId);
ensure(ev.hasReservation, "scale-out effect missing reservation");
tgt.reservationCount--;
tgt.activeNodeCount++;
nd.active = true;
nd.initialized = true;
nd.cpu = ev.newCpu;
nd.mem = ev.newMem;
nd.pending = false;
nd.pendingEffectTime = -1;
Instance& outInst = in.instances[instanceIndex[ev.instanceId]];
outInst.activeNodes++;
};
// --- Initialization round (t = 0).
long long a0ll = nonNegativeInt(user_out, "a0");
ensure(a0ll <= INT_MAX, "a0 too large");
int a0 = static_cast<int>(a0ll);
for (int i = 0; i < a0; ++i) {
int serverId = static_cast<int>(user_out.readInt(0, MAX_ID, "server_id"));
int modelId = static_cast<int>(user_out.readInt(1, MAX_ID, "model_id"));
ensure(!chosenIds.count(serverId), "chosen identifier reused");
chosenIds.insert(serverId);
ensure(!servers.count(serverId) || !servers[serverId].exists, "server id already exists");
const ServerModel& model = getModel(modelId);
ServerState st;
st.id = serverId;
st.modelId = modelId;
st.capCpu = model.cpu;
st.capMem = model.mem;
st.costPerSecond = model.cost;
st.exists = true;
servers[serverId] = st;
runningCostPerRound += static_cast<long double>(st.costPerSecond);
}
long long p0ll = nonNegativeInt(user_out, "p0");
ensure(p0ll <= INT_MAX, "p0 too large");
int p0 = static_cast<int>(p0ll);
unordered_set<int> placed;
placed.reserve(nodes.size() * 2 + 1);
for (int i = 0; i < p0; ++i) {
int nodeId = static_cast<int>(user_out.readInt(0, MAX_ID, "node_id"));
int serverId = static_cast<int>(user_out.readInt(0, MAX_ID, "server_id"));
ensure(nodes.count(nodeId), "initial placement references unknown node");
NodeState& nd = nodes[nodeId];
ensure(!nd.initialized, "node placed more than once");
ensure(servers.count(serverId) && servers[serverId].exists, "placement references unknown server");
ServerState& srv = servers[serverId];
ensure(serverCanFit(srv, in.cpuMin, in.memMin), "initial placement exceeds server capacity");
nd.initialized = true;
nd.active = true;
nd.serverId = serverId;
nd.cpu = in.cpuMin;
nd.mem = in.memMin;
srv.usedCpu += in.cpuMin;
srv.usedMem += in.memMin;
srv.activeNodeCount++;
in.instances[instanceIndex[nd.instanceId]].activeNodes++;
placed.insert(nodeId);
}
ensure(static_cast<int>(placed.size()) == static_cast<int>(nodes.size()),
"all initial nodes must be placed exactly once");
for (const auto& inst : in.instances) {
ensure(inst.activeNodes == inst.b, "instance active node count after init is invalid");
}
user_in << 0 << '\n';
user_in.flush();
// --- Main rounds.
for (int t = 1; t <= T_ROUNDS; ++t) {
// Apply all delayed effects that take place at this round.
if (t < static_cast<int>(events.size())) {
for (const PendingEvent& ev : events[t]) {
applyEventAtRound(ev);
}
events[t].clear();
}
// Send arrivals for current round.
const auto& arrivals = in.arrivalsByRound[t];
user_in << arrivals.size() << '\n';
for (const auto& rg : arrivals) {
user_in << rg.instanceId << ' ' << rg.queryId << ' ' << rg.qps << '\n';
}
user_in.flush();
// Add arrivals to middleware now (they are available immediately at this round).
for (const auto& rg : arrivals) {
addMiddleware(rg.instanceId, rg.queryId, t, rg.qps);
}
// Read action block header.
long long cntNewLL = nonNegativeInt(user_out, "cnt_new");
long long cntDelLL = nonNegativeInt(user_out, "cnt_del");
long long cntLocalLL = nonNegativeInt(user_out, "cnt_local");
long long cntCrossLL = nonNegativeInt(user_out, "cnt_cross");
long long cntOutLL = nonNegativeInt(user_out, "cnt_out");
ensure(cntNewLL <= INT_MAX && cntDelLL <= INT_MAX && cntLocalLL <= INT_MAX &&
cntCrossLL <= INT_MAX && cntOutLL <= INT_MAX,
"operation count too large");
int cntNew = static_cast<int>(cntNewLL);
int cntDel = static_cast<int>(cntDelLL);
int cntLocal = static_cast<int>(cntLocalLL);
int cntCross = static_cast<int>(cntCrossLL);
int cntOut = static_cast<int>(cntOutLL);
// NEW_SERVER.
for (int i = 0; i < cntNew; ++i) {
int serverId = static_cast<int>(user_out.readInt(0, MAX_ID, "server_id"));
int modelId = static_cast<int>(user_out.readInt(1, MAX_ID, "model_id"));
ensure(!chosenIds.count(serverId), "chosen identifier reused");
chosenIds.insert(serverId);
ensure(!servers.count(serverId) || !servers[serverId].exists, "server id already exists");
const ServerModel& model = getModel(modelId);
ServerState st;
st.id = serverId;
st.modelId = modelId;
st.capCpu = model.cpu;
st.capMem = model.mem;
st.costPerSecond = model.cost;
st.exists = true;
servers[serverId] = st;
runningCostPerRound += static_cast<long double>(st.costPerSecond);
}
// DEL_SERVER.
for (int i = 0; i < cntDel; ++i) {
int serverId = static_cast<int>(user_out.readInt(0, MAX_ID, "server_id"));
ensure(servers.count(serverId) && servers[serverId].exists, "deleting unknown server");
ServerState& srv = servers[serverId];
ensure(srv.activeNodeCount == 0 && srv.reservationCount == 0 &&
srv.usedCpu == 0 && srv.usedMem == 0,
"attempt to delete non-empty server");
runningCostPerRound -= static_cast<long double>(srv.costPerSecond);
srv.exists = false;
}
// LOCAL_SCALE.
for (int i = 0; i < cntLocal; ++i) {
int nodeId = static_cast<int>(user_out.readInt(0, MAX_ID, "node_id"));
long long dCpu = user_out.readInt(-1000000000LL, 1000000000LL, "delta_cpu");
long long dMem = user_out.readInt(-1000000000LL, 1000000000LL, "delta_mem");
NodeState& nd = ensureNodeExists(nodeId);
ensure(nd.initialized && nd.active, "LOCAL_SCALE targets inactive or unknown node");
ensure(!nd.pending, "node has pending operation");
ensure(!(dCpu == 0 && dMem == 0), "LOCAL_SCALE no-op is invalid");
ensure(isMultiple(dCpu, in.gCpu), "LOCAL_SCALE delta cpu violates granularity");
ensure(isMultiple(dMem, in.gMem), "LOCAL_SCALE delta mem violates granularity");
long long newCpu = nd.cpu + dCpu;
long long newMem = nd.mem + dMem;
ensure(validateFinalSpec(newCpu, newMem, true), "LOCAL_SCALE resulting spec is invalid");
if (newCpu == 0) {
ensure(!nd.primary, "primary node cannot be released");
}
nd.pending = true;
nd.pendingEffectTime = t + in.Tlu;
PendingEvent ev;
ev.type = OpType::LOCAL;
ev.nodeId = nodeId;
ev.effectTime = nd.pendingEffectTime;
ev.newCpu = newCpu;
ev.newMem = newMem;
if (ev.effectTime < static_cast<int>(events.size())) {
events[ev.effectTime].push_back(ev);
}
}
// CROSS_SCALE.
for (int i = 0; i < cntCross; ++i) {
int nodeId = static_cast<int>(user_out.readInt(0, MAX_ID, "node_id"));
int targetServerId = static_cast<int>(user_out.readInt(0, MAX_ID, "target_server_id"));
long long dCpu = user_out.readInt(-1000000000LL, 1000000000LL, "delta_cpu");
long long dMem = user_out.readInt(-1000000000LL, 1000000000LL, "delta_mem");
NodeState& nd = ensureNodeExists(nodeId);
ensure(nd.initialized && nd.active, "CROSS_SCALE targets inactive or unknown node");
ensure(!nd.pending, "node has pending operation");
ensure(servers.count(targetServerId) && servers[targetServerId].exists,
"CROSS_SCALE target server does not exist");
ensure(targetServerId != nd.serverId, "CROSS_SCALE requires different target server");
ensure(isMultiple(dCpu, in.gCpu), "CROSS_SCALE delta cpu violates granularity");
ensure(isMultiple(dMem, in.gMem), "CROSS_SCALE delta mem violates granularity");
long long newCpu = nd.cpu + dCpu;
long long newMem = nd.mem + dMem;
ensure(validateFinalSpec(newCpu, newMem, true), "CROSS_SCALE resulting spec is invalid");
if (newCpu == 0) {
ensure(!nd.primary, "primary node cannot be released");
}
bool reserve = (newCpu > 0);
if (reserve) {
ServerState& tgt = servers[targetServerId];
ensure(serverCanFit(tgt, newCpu, newMem),
"CROSS_SCALE reservation exceeds target server capacity");
tgt.usedCpu += newCpu;
tgt.usedMem += newMem;
tgt.reservationCount++;
}
nd.pending = true;
nd.pendingEffectTime = t + in.Tcu;
PendingEvent ev;
ev.type = OpType::CROSS;
ev.nodeId = nodeId;
ev.effectTime = nd.pendingEffectTime;
ev.newCpu = newCpu;
ev.newMem = newMem;
ev.targetServerId = targetServerId;
ev.hasReservation = reserve;
if (ev.effectTime < static_cast<int>(events.size())) {
events[ev.effectTime].push_back(ev);
}
}
// SCALE_OUT.
for (int i = 0; i < cntOut; ++i) {
int instanceId = static_cast<int>(user_out.readInt(1, MAX_ID, "instance_id"));
int nodeId = static_cast<int>(user_out.readInt(0, MAX_ID, "target_node_id"));
int targetServerId = static_cast<int>(user_out.readInt(0, MAX_ID, "target_server_id"));
long long cpu = user_out.readInt(0, 1000000000LL, "cpu");
long long mem = user_out.readInt(0, 1000000000LL, "mem");
ensure(instanceIndex.count(instanceId), "SCALE_OUT unknown instance id");
ensure(servers.count(targetServerId) && servers[targetServerId].exists,
"SCALE_OUT target server does not exist");
ensure(!chosenIds.count(nodeId), "chosen identifier reused");
chosenIds.insert(nodeId);
ensure(!nodes.count(nodeId), "SCALE_OUT node id already exists");
ensure(validateFinalSpec(cpu, mem, false), "SCALE_OUT node spec invalid");
ensure(isMultiple(cpu, in.gCpu), "SCALE_OUT cpu violates granularity");
ensure(isMultiple(mem, in.gMem), "SCALE_OUT mem violates granularity");
ServerState& tgt = servers[targetServerId];
ensure(serverCanFit(tgt, cpu, mem), "SCALE_OUT reservation exceeds target server capacity");
tgt.usedCpu += cpu;
tgt.usedMem += mem;
tgt.reservationCount++;
NodeState nd;
nd.id = nodeId;
nd.instanceId = instanceId;
nd.primary = false;
nd.initialized = true;
nd.active = false;
nd.serverId = targetServerId;
nd.cpu = 0;
nd.mem = 0;
nd.pending = true;
nd.pendingEffectTime = t + in.Tso;
nodes[nodeId] = nd;
PendingEvent ev;
ev.type = OpType::SCALE_OUT;
ev.nodeId = nodeId;
ev.effectTime = nd.pendingEffectTime;
ev.newCpu = cpu;
ev.newMem = mem;
ev.targetServerId = targetServerId;
ev.hasReservation = true;
ev.instanceId = instanceId;
if (ev.effectTime < static_cast<int>(events.size())) {
events[ev.effectTime].push_back(ev);
}
}
// Running cost for this round is counted after immediate server operations.
C_r += runningCostPerRound;
// Dispatches.
long long dTll = nonNegativeInt(user_out, "d_t");
ensure(dTll <= INT_MAX, "d_t too large");
int dT = static_cast<int>(dTll);
for (int i = 0; i < dT; ++i) {
int instanceId = static_cast<int>(user_out.readInt(1, MAX_ID, "instance_id"));
int queryId = static_cast<int>(user_out.readInt(1, MAX_ID, "query_id"));
int nodeId = static_cast<int>(user_out.readInt(0, MAX_ID, "node_id"));
long long amount = user_out.readInt(1, LLONG_MAX / 4, "amount");
ensure(instanceIndex.count(instanceId), "dispatch unknown instance id");
ensure(queryIndex.count(queryId), "dispatch unknown query id");
ensure(nodes.count(nodeId), "dispatch unknown node id");
NodeState& nd = nodes[nodeId];
ensure(nd.active, "dispatch to inactive node");
ensure(nd.instanceId == instanceId, "dispatch node does not belong to instance");
const Query& q = in.queries[queryIndex[queryId]];
if (q.type == 0) {
const Instance& inst = in.instances[instanceIndex[instanceId]];
ensure(nodeId == inst.primaryNodeId, "write query must be dispatched to primary node");
}
vector<MiddlewareChunk> taken = consumeMiddleware(instanceId, queryId, amount);
for (const auto& chunk : taken) {
enqueueToNode(nd, instanceId, queryId, chunk.arrivalTime, chunk.count);
}
}
// Simulate one second of execution for active nodes.
for (auto& [nodeId, nd] : nodes) {
if (!nd.active) continue;
if (nd.cpu <= 0) continue;
long long budget = nd.cpu;
while (budget > 0 && !nd.queue.empty()) {
QueueChunk& front = nd.queue.front();
const Query& q = in.queries[queryIndex[front.queryId]];
if (nd.headRemainingCpu == 0) {
nd.headRemainingCpu = q.cpu;
}
bool started = (nd.headRemainingCpu < q.cpu);
if (!started && q.mem > nd.mem) {
break;
}
if (budget >= nd.headRemainingCpu) {
budget -= nd.headRemainingCpu;
nd.headRemainingCpu = 0;
const Instance& inst = in.instances[instanceIndex[front.instanceId]];
long long overloadTime = static_cast<long long>(t) - front.arrivalTime;
long long extra = max(0LL, overloadTime - q.qto);
long long penalty = static_cast<long long>(in.weights[inst.level - 1]) * extra;
C_s += static_cast<long double>(penalty);
front.count--;
if (front.count == 0) {
nd.queue.pop_front();
}
} else {
nd.headRemainingCpu -= budget;
budget = 0;
}
}
if (nd.queue.empty()) nd.headRemainingCpu = 0;
}
user_in << 0 << '\n';
user_in.flush();
}
user_out.readEof();
// Final completeness checks.
ensure(middleware.empty(), "not all arrived requests were dispatched");
for (const auto& [nodeId, nd] : nodes) {
ensure(nd.queue.empty() && nd.headRemainingCpu == 0,
"not all dispatched requests were fully executed by the end");
}
if (score_log.is_open()) {
score_log << "HOTDBSCORE_V1\n";
score_log << fixed << setprecision(20) << C_r << ' ' << C_s << '\n';
}
return 0;
}
