void software_free(gpointer data) {
	Software* software = data;
	MAGIC_ASSERT(software);

	g_string_free(software->arguments, TRUE);
	g_string_free(software->pluginPath, TRUE);

	MAGIC_CLEAR(software);
	g_free(software);
}

void scheduler_unref(Scheduler* scheduler) {
    MAGIC_ASSERT(scheduler);
    g_mutex_lock(&(scheduler->globalLock));
    scheduler->referenceCount--;
    gboolean shouldFree = (scheduler->referenceCount <= 0) ? TRUE : FALSE;
    g_mutex_unlock(&(scheduler->globalLock));
    if(shouldFree) {
        _scheduler_free(scheduler);
    }
}

gsize socket_getOutputBufferSpace(Socket* socket) {
    MAGIC_ASSERT(socket);
    utility_assert(socket->outputBufferSize >= socket->outputBufferLength);
    gsize bufferSize = socket_getOutputBufferSize(socket);
    if(bufferSize < socket->outputBufferLength) {
        return 0;
    } else {
        return bufferSize - socket->outputBufferLength;
    }
}

static void _address_free(Address* address) {
    MAGIC_ASSERT(address);

    g_free(address->ipString);
    g_free(address->name);
    g_free(address->idString);

    MAGIC_CLEAR(address);
    g_free(address);
}

gpointer engine_get(Engine* engine, EngineStorage type, GQuark id) {
	MAGIC_ASSERT(engine);

	/*
	 * Return the item corresponding to type and id in a thread-safe way.
	 * I believe for now no protections are necessary since our registry
	 * is read-only.
	 */
	return registry_get(engine->registry, type, &id);
}

void tracker_addVirtualProcessingDelay(Tracker* tracker, SimulationTime delay) {
    MAGIC_ASSERT(tracker);

    if(_tracker_getFlags(tracker) & TRACKER_FLAGS_NODE) {
        (tracker->numDelayedTotal)++;
        tracker->delayTimeTotal += delay;
        (tracker->numDelayedLastInterval)++;
        tracker->delayTimeLastInterval += delay;
    }
}

void connectnetwork_run(ConnectNetworkAction* action) {
    MAGIC_ASSERT(action);

    internetwork_connectNetworks(worker_getInternet(),
                                 action->sourceClusterID, action->destinationClusterID,
                                 action->latency, action->jitter, action->packetloss,
                                 action->latencymin, action->latencyQ1, action->latencymean,
                                 action->latencyQ3, action->latencymax);

}

SimulationTime cpu_getDelay(CPU* cpu) {
	MAGIC_ASSERT(cpu);

	/* we only have delay if we've crossed the threshold */
	SimulationTime builtUpDelay = cpu->timeCPUAvailable - cpu->now;
	if(builtUpDelay > cpu->threshold) {
		return builtUpDelay;
	}
	return 0;
}

void engine_put(Engine* engine, EngineStorage type, GQuark* id, gpointer item) {
	MAGIC_ASSERT(engine);

	/*
	 * put the item corresponding to type and id in a thread-safe way.
	 * I believe for now no protections are necessary since our registry
	 * is filled before simulation and is read only.
	 */
	registry_put(engine->registry, type, id, item);
}

static gint _host_monitorDescriptor(Host* host, Descriptor* descriptor) {
	MAGIC_ASSERT(host);

	/* make sure there are no collisions before inserting */
	gint* handle = descriptor_getHandleReference(descriptor);
	utility_assert(handle && !host_lookupDescriptor(host, *handle));
	g_hash_table_replace(host->descriptors, handle, descriptor);

	return *handle;
}

static gsize _tcp_getBufferSpaceOut(TCP* tcp) {
	MAGIC_ASSERT(tcp);
	/* account for throttled and retransmission buffer */
	gssize s = (gssize)(socket_getOutputBufferSpace(&(tcp->super)) - tcp->throttledOutputLength - tcp->retransmissionLength);
	gsize space = MAX(0, s);
	if(space == 0) {
		descriptor_adjustStatus((Descriptor*)tcp, DS_WRITABLE, FALSE);
	}
	return space;
}

void epoll_tryNotify(Epoll* epoll) {
	MAGIC_ASSERT(epoll);

	if(_epoll_isReadyToNotify(epoll)) {
		application_notify(epoll->ownerApplication);

		/* check if we need to be notified again */
		_epoll_ensureTriggers(epoll);
	}
}

static void _tcp_endOfFileSignalled(TCP* tcp) {
	MAGIC_ASSERT(tcp);

	debug("%s <-> %s: signaling close to user, socket no longer usable", tcp->super.boundString, tcp->super.peerString);
	tcp->flags |= TCPF_EOF_SIGNALED;

	/* user can no longer access socket */
	descriptor_adjustStatus(&(tcp->super.super.super), DS_CLOSED, TRUE);
	descriptor_adjustStatus(&(tcp->super.super.super), DS_ACTIVE, FALSE);
}

void tracker_free(Tracker* tracker) {
    MAGIC_ASSERT(tracker);

    g_hash_table_foreach(tracker->allocatedLocations, _tracker_freeAllocatedLocations, NULL);
    g_hash_table_destroy(tracker->allocatedLocations);
    g_hash_table_destroy(tracker->socketStats);

    MAGIC_CLEAR(tracker);
    g_free(tracker);
}

void worker_free(gpointer data) {
	Worker* worker = data;
	MAGIC_ASSERT(worker);

	/* calls the destroy functions we specified in g_hash_table_new_full */
	g_hash_table_destroy(worker->plugins);

	MAGIC_CLEAR(worker);
	g_free(worker);
}

void udp_processPacket(UDP* udp, Packet* packet) {
	MAGIC_ASSERT(udp);

	/* UDP packet contains data for user and can be buffered immediately */
	if(packet_getPayloadLength(packet) > 0) {
		if(!socket_addToInputBuffer((Socket*)udp, packet)) {
			packet_addDeliveryStatus(packet, PDS_RCV_SOCKET_DROPPED);
		}
	}
}

Event* scheduler_pop(Scheduler* scheduler) {
    MAGIC_ASSERT(scheduler);

    /* this function should block until a non-null event is available for the worker to run.
     * return NULL only to signal the worker thread to quit */

    while(scheduler->isRunning) {
        /* pop from a queue based on the policy */
        Event* nextEvent = scheduler->policy->pop(scheduler->policy, scheduler->currentRound.endTime);

        if(nextEvent != NULL) {
            /* we have an event, let the worker run it */
            return nextEvent;
        } else if(scheduler->policyType == SP_SERIAL_GLOBAL) {
            /* the running thread has no more events to execute this round, but we only have a
             * single, global, serial queue, so returning NULL without blocking is OK. */
            return NULL;
        } else {
            /* the running thread has no more events to execute this round and we need to block it
             * so that we can wait for all threads to finish events from this round. We want to
             * track idle times, so let's start by making sure we have timer elements in place. */
            GTimer* executeEventsBarrierWaitTime = g_hash_table_lookup(scheduler->threadToWaitTimerMap, GUINT_TO_POINTER(pthread_self()));

            /* wait for all other worker threads to finish their events too, and track wait time */
            if(executeEventsBarrierWaitTime) {
                g_timer_continue(executeEventsBarrierWaitTime);
            }
            countdownlatch_countDownAwait(scheduler->executeEventsBarrier);
            if(executeEventsBarrierWaitTime) {
                g_timer_stop(executeEventsBarrierWaitTime);
            }

            /* now all threads reached the current round end barrier time.
             * asynchronously collect some stats that the main thread will use. */
            if(scheduler->policy->getNextTime) {
                SimulationTime nextTime = scheduler->policy->getNextTime(scheduler->policy);
                g_mutex_lock(&(scheduler->globalLock));
                scheduler->currentRound.minNextEventTime = MIN(scheduler->currentRound.minNextEventTime, nextTime);
                g_mutex_unlock(&(scheduler->globalLock));
            }

            /* clear all log messages from the last round */
            logger_flushRecords(logger_getDefault(), pthread_self());

            /* wait for other threads to finish their collect step */
            countdownlatch_countDownAwait(scheduler->collectInfoBarrier);

            /* now wait for main thread to process a barrier update for the next round */
            countdownlatch_countDownAwait(scheduler->prepareRoundBarrier);
        }
    }

    /* scheduler is done, return NULL to stop worker */
    return NULL;
}

void socket_setOutputBufferSize(Socket* socket, gsize newSize) {
    MAGIC_ASSERT(socket);
    if(newSize >= socket->outputBufferLength) {
        socket->outputBufferSize = newSize;
        socket->outputBufferSizePending = 0;
    } else {
        /* ensure positive size, reduce size as buffer drains */
        socket->outputBufferSize = socket->outputBufferLength;
        socket->outputBufferSizePending = newSize;
    }
}

static void _epollwatch_free(gpointer data) {
	EpollWatch* watch = data;
	MAGIC_ASSERT(watch);

	descriptor_removeStatusListener(watch->descriptor, watch->listener);
	listener_free(watch->listener);
	descriptor_unref(watch->descriptor);

	MAGIC_CLEAR(watch);
	g_free(watch);
}

void notifyplugin_run(NotifyPluginEvent* event, Host* node) {
	MAGIC_ASSERT(event);

	debug("event started");

	/* check in with epoll to make sure we should carry out the notification */
	Epoll* epoll = (Epoll*) host_lookupDescriptor(node, event->epollHandle);
	epoll_tryNotify(epoll);

	debug("event finished");
}