Merge branch 'db-field-expiry'

Author: jvde-github

Source/Application/Engine.cpp

@@ -202,6 +202,10 @@ void Engine::run(WebViewer *viewer, ControlCore *control)
 	}
 
 	const int SLEEP = 50;
+#ifdef HASWEBVIEWER
+	const int TICK_INTERVAL = 60;
+	int tick_countdown = 0;
+#endif
 	auto time_start = high_resolution_clock::now();
 	auto time_timeout_start = time_start;
 	auto time_last = time_start;
@@ -224,6 +228,18 @@ void Engine::run(WebViewer *viewer, ControlCore *control)
 		if (iscallback)
 			std::this_thread::sleep_for(std::chrono::milliseconds(SLEEP));
 
+#ifdef HASWEBVIEWER
+		// above the verbose/timeout shortcut below or it never runs by default
+		if (--tick_countdown <= 0)
+		{
+			tick_countdown = TICK_INTERVAL * 1000 / SLEEP;
+
+			std::time_t tick_now = std::time(nullptr);
+			for (auto v : viewers)
+				v->tick(tick_now);
+		}
+#endif
+
 		if (!oneverbose && !timeout)
 			continue;
 

Source/Application/WebViewer.cpp

@@ -511,6 +511,14 @@ void WebViewer::attachEngine(AIS::Model &model, Connection<JSON::JSON> &json, De
 	endAttach();
 }
 
+void WebViewer::tick(std::time_t now)
+{
+	std::lock_guard<std::recursive_mutex> lock(state_mtx);
+
+	for (auto &s : states)
+		s->tick(now);
+}
+
 void WebViewer::resetStatistics()
 {
 	std::lock_guard<std::recursive_mutex> lock(state_mtx);
@@ -1082,6 +1090,9 @@ Setting &WebViewer::SetKey(AIS::Keys key, const std::string &arg)
 	case AIS::KEY_SETTING_CUTOFF:
 		settings.tracking.cutoff = Util::Parse::Integer(arg, 0, 10000);
 		break;
+	case AIS::KEY_SETTING_EXPIRE:
+		settings.tracking.expire_fields = Util::Parse::Switch(arg);
+		break;
 	case AIS::KEY_SETTING_SHARE_LOC:
 		settings.tracking.latlon_share = Util::Parse::Switch(arg);
 		frontend.setShareLoc(settings.tracking.latlon_share);

Source/Application/WebViewer.h

@@ -231,6 +231,9 @@ class WebViewer : public IO::HTTPServer, public Setting
 	void stopServing();
 	// for good: unbinds, writes the statistics file and joins the server thread
 	void shutdown();
+	// periodic maintenance from the engine loop; handlers gate on the timestamp
+	// and must return promptly, no blocking I/O
+	void tick(std::time_t now);
 	// drop the accumulated statistics, keeping the configuration (Android only)
 	void resetStatistics();
 	// Back to defaults, then the one setting the config file does not own. Taking

Source/JSON/KeyDefs.h

@@ -112,6 +112,7 @@ X(KEY_SETTING_GPS, "", "", "", "", "gps", "", "", "", nullptr)
 X(KEY_SETTING_GRDB, "", "", "", "", "grdb", "", "", "", nullptr)
 X(KEY_SETTING_GROUPS_IN, "", "", "", "", "groups_in", "", "", "", nullptr)
 X(KEY_SETTING_GZIP, "", "", "", "", "gzip", "", "", "", nullptr)
+X(KEY_SETTING_EXPIRE, "", "", "", "", "expire", "", "", "", nullptr)
 X(KEY_SETTING_SSL_VERIFY, "", "", "", "", "ssl_verify", "", "", "", nullptr)
 X(KEY_SETTING_HACKRF, "", "", "", "", "hackrf", "", "", "", nullptr)
 X(KEY_SETTING_HISTORY, "", "", "", "", "history", "", "", "", nullptr)

Source/Tracking/DB.cpp

@@ -39,6 +39,8 @@ void DB::setup()
 	paths.resize(Npaths);
 	hash_table.resize(HASH_SIZE);
 
+	last_sweep = std::time(nullptr);
+
 	first = Nships - 1;
 	last = 0;
 	count = 0;
@@ -839,7 +841,8 @@ bool DB::updateFields(const JSON::Member &p, const AIS::Message *msg, Ship &v, b
 		v.received_stations = p.Get().getInt();
 		break;
 	case AIS::KEY_ALT:
-		v.altitude = p.Get().getInt();
+		if (msg->type() == 9)
+			v.altitude = p.Get().getInt();
 		break;
 	case AIS::KEY_VIRTUAL_AID:
 		v.setVirtualAid(p.Get().getBool());
@@ -991,7 +994,7 @@ bool DB::updateShip(const JSON::JSON &data, TAG &tag, Ship &ship)
 
 	ship.ppm = tag.ppm;
 	ship.level = tag.level;
-	ship.msg_type |= 1 << type;
+	ship.markType(type);
 
 	if (msg->getChannel() >= 'A' && msg->getChannel() <= 'D')
 		ship.orOpChannels(1 << (msg->getChannel() - 'A'));
@@ -1029,11 +1032,13 @@ bool DB::updateShip(const JSON::JSON &data, TAG &tag, Ship &ship)
 	return positionUpdated;
 }
 
-static bool isBinaryContentKey(int key)
+static bool isBinaryContent(const JSON::Member &p)
 {
-	switch (key)
+	switch (p.Key())
 	{
 	case AIS::KEY_TEXT:
+		// getText trims the '@'/space padding, so a pure-padding broadcast is empty
+		return !p.Get().getString().empty();
 	case AIS::KEY_CREW_COUNT:
 	case AIS::KEY_PASSENGER_COUNT:
 	case AIS::KEY_SHIPBOARD_PERSONNEL_COUNT:
@@ -1109,7 +1114,7 @@ void DB::processBinaryMessage(const JSON::JSON &data, Ship &ship, bool &position
 		{
 			loc_lon = p.Get().getFloat();
 		}
-		else if (isBinaryContentKey(p.Key()))
+		else if (isBinaryContent(p))
 		{
 			has_content = true;
 		}
@@ -1269,6 +1274,23 @@ void DB::Receive(const JSON::JSON *data, int len, TAG &tag)
 	Send(data, len, tag);
 }
 
+void DB::tick(std::time_t now)
+{
+	if (!expire_fields)
+		return;
+
+	std::lock_guard<std::mutex> lock(mtx);
+
+	if (now - last_sweep < TIME_HISTORY)
+		return;
+
+	last_sweep = now;
+
+	for (int ptr = first; ptr != -1; ptr = ships[ptr].incoming.next)
+		if (ships[ptr].mmsi)
+			ships[ptr].decayAndExpire();
+}
+
 bool DB::Save(std::ofstream &file)
 {
 	std::lock_guard<std::mutex> lock(mtx);
@@ -1359,6 +1381,9 @@ bool DB::Load(std::ifstream &file)
 		// Not persisted; treat all loaded ships as having static data
 		temp_ships[i].last_static_signal = temp_ships[i].last_signal;
 
+		// Not persisted; stale records are left to the sweep
+		temp_ships[i].type_ttl = temp_ships[i].msg_type;
+
 		if (i > 0 && temp_ships[i].last_signal < previous_signal)
 		{
 			Error() << "DB: Ships not in chronological order at index " << i;

Source/Tracking/DB.h

@@ -81,6 +81,9 @@ class DB : public StreamIn<JSON::JSON>,
 	int Npaths = Nships * 16;
 	int HASH_SIZE = 8209;
 
+	bool expire_fields = false;
+	std::time_t last_sweep = 0;
+
 	struct HashBucket
 	{
 		int first = -1;
@@ -133,7 +136,9 @@ class DB : public StreamIn<JSON::JSON>,
 	std::mutex mtx;
 
 	void setup();
+	void tick(std::time_t now);
 	void setTimeHistory(int t) { TIME_HISTORY = t; }
+	void setExpireFields(bool b) { expire_fields = b; }
 	void setShareLatLon(bool b) { latlon_share = b; }
 	bool getShareLatLon() { return latlon_share; }
 

Source/Tracking/ReceiverTracker.cpp

@@ -26,6 +26,7 @@ void ReceiverTracker::applyConfig(const TrackingConfig &cfg, const AIS::Filter &
 	ships.setMsgSave(cfg.msg_save);
 	ships.setOwnMMSI(cfg.own_mmsi);
 	ships.setTimeHistory(cfg.time_history);
+	ships.setExpireFields(cfg.expire_fields);
 	ships.setFilter(f);
 
 	// applied unconditionally: a config that drops "cutoff" must go back to the

Source/Tracking/ReceiverTracker.h

@@ -44,6 +44,7 @@ struct TrackingConfig
 	bool use_GPS = true;
 	uint32_t own_mmsi = 0;
 	int time_history = 30 * 60;
+	bool expire_fields = false;
 	// 0 means "not configured": LONG_RANGE_CUTOFF_DEFAULT is used instead
 	int cutoff = 0;
 };
@@ -106,6 +107,8 @@ struct ReceiverTracker
 	void writeSummary(std::ostream &out);
 
 	// Ship data queries
+	void tick(std::time_t now) { ships.tick(now); }
+
 	int getCount() { return ships.getCount(); }
 	int getMaxCount() { return ships.getMaxCount(); }
 	float getMsgRate() { return hist_second.getAverage(); }

Source/Tracking/Ships.cpp

@@ -28,6 +28,7 @@ void Ship::reset()
 	path_ptr = -1;
 
 	mmsi = count = msg_type = shiptype = group_mask = 0;
+	type_ttl = 0;
 	flags.reset();
 
 	heading = HEADING_UNDEFINED;
@@ -68,6 +69,130 @@ void Ship::reset()
 	msg.clear();
 }
 
+// Which fields each message type can refresh, indexed by type; F_SIGNAL is in every
+// row as reception data comes with every message. A missing field here expires
+// early, a spurious one never expires.
+static const uint32_t TYPE_FIELDS[MAX_MSG_TYPE + 1] = {
+	0,																			// 0  does not exist
+	F_SIGNAL | F_LATLON | F_SPEED_COG | F_HEADING | F_STATUS | F_MANEUVER | F_RECV_STATIONS, // 1  position report
+	F_SIGNAL | F_LATLON | F_SPEED_COG | F_HEADING | F_STATUS | F_MANEUVER | F_RECV_STATIONS, // 2  position report
+	F_SIGNAL | F_LATLON | F_SPEED_COG | F_HEADING | F_STATUS | F_MANEUVER | F_RECV_STATIONS, // 3  position report
+	F_SIGNAL | F_LATLON | F_RECV_STATIONS,										// 4  base station report
+	F_SIGNAL | F_VOYAGE | F_STATIC,												// 5  static and voyage
+	F_SIGNAL | F_LATLON | F_OFF_POSITION | F_STATIC | F_VOYAGE,					// 6  addressed binary, buoy/AtoN monitor payloads
+	F_SIGNAL,																	// 7  binary acknowledge
+	F_SIGNAL | F_STATIC | F_VOYAGE | F_OFF_POSITION,							// 8  broadcast binary, inland static and AtoN monitor payloads
+	F_SIGNAL | F_LATLON | F_SPEED_COG | F_ALTITUDE,								// 9  SAR aircraft
+	F_SIGNAL,																	// 10 UTC inquiry
+	F_SIGNAL | F_LATLON | F_RECV_STATIONS,										// 11 UTC response
+	F_SIGNAL,																	// 12 addressed safety
+	F_SIGNAL,																	// 13 safety acknowledge
+	F_SIGNAL,																	// 14 broadcast safety
+	F_SIGNAL,																	// 15 interrogation
+	F_SIGNAL,																	// 16 assignment
+	F_SIGNAL,																	// 17 DGNSS broadcast
+	F_SIGNAL | F_LATLON | F_SPEED_COG | F_HEADING | F_COMM_CAP,					// 18 class B position
+	F_SIGNAL | F_LATLON | F_SPEED_COG | F_HEADING | F_STATIC,					// 19 class B extended
+	F_SIGNAL,																	// 20 link management
+	F_SIGNAL | F_LATLON | F_OFF_POSITION | F_STATIC,							// 21 aid to navigation
+	F_SIGNAL,																	// 22 channel management
+	F_SIGNAL,																	// 23 group assignment
+	F_SIGNAL | F_STATIC,														// 24 class B static
+	F_SIGNAL | F_LATLON | F_OFF_POSITION | F_STATIC | F_VOYAGE,					// 25 single slot binary, same payloads as 6/8
+	F_SIGNAL | F_LATLON | F_OFF_POSITION | F_STATIC | F_VOYAGE | F_RECV_STATIONS, // 26 multiple slot binary, same payloads as 6/8
+	F_SIGNAL | F_LATLON | F_SPEED_COG | F_STATUS,								// 27 long range
+	F_SIGNAL | F_LATLON | F_STATIC,												// 28 AtoN report
+};
+
+void Ship::decayAndExpire()
+{
+	if (~type_ttl & msg_type)
+	{
+		uint32_t supported = 0;
+		for (int t = 1; t <= MAX_MSG_TYPE; t++)
+			if (type_ttl & (1 << t))
+				supported |= TYPE_FIELDS[t];
+
+		clearFields(~supported);
+
+		msg_type = type_ttl;
+		setType();
+	}
+
+	type_ttl = 0;
+}
+
+void Ship::clearFields(uint32_t doomed)
+{
+	if (doomed & F_LATLON)
+	{
+		lat = LAT_UNDEFINED;
+		lon = LON_UNDEFINED;
+		distance = DISTANCE_UNDEFINED;
+		angle = ANGLE_UNDEFINED;
+		setApproximate(0);
+		setValidated(0);
+		setRAIM(0);
+		setAssigned(0);
+	}
+	if (doomed & F_SPEED_COG)
+	{
+		speed = SPEED_UNDEFINED;
+		cog = COG_UNDEFINED;
+	}
+	if (doomed & F_HEADING)
+		heading = HEADING_UNDEFINED;
+	if (doomed & F_STATUS)
+		status = STATUS_UNDEFINED;
+	if (doomed & F_MANEUVER)
+		setManeuver(0);
+	if (doomed & F_ALTITUDE)
+		altitude = ALT_UNDEFINED;
+	if (doomed & F_RECV_STATIONS)
+		received_stations = RECEIVED_STATIONS_UNDEFINED;
+	if (doomed & F_OFF_POSITION)
+		setOffPosition(0);
+	if (doomed & F_VOYAGE)
+	{
+		memset(destination, 0, sizeof(destination));
+		month = ETA_MONTH_UNDEFINED;
+		day = ETA_DAY_UNDEFINED;
+		hour = ETA_HOUR_UNDEFINED;
+		minute = ETA_MINUTE_UNDEFINED;
+		draught = DRAUGHT_UNDEFINED;
+	}
+	if (doomed & F_STATIC)
+	{
+		memset(shipname, 0, sizeof(shipname));
+		memset(callsign, 0, sizeof(callsign));
+		memset(vendorid, 0, sizeof(vendorid));
+		memset(vin, 0, sizeof(vin));
+		unit_model = unit_serial = -1;
+		IMO = IMO_UNDEFINED;
+		shiptype = 0;
+		to_port = to_bow = to_starboard = to_stern = DIMENSION_UNDEFINED;
+		setDTE(0);
+		setVirtualAid(0);
+	}
+	if (doomed & F_COMM_CAP)
+	{
+		setCSUnit(0);
+		setDisplay(0);
+		setDSC(0);
+		setBand(0);
+		setMsg22(0);
+	}
+	if (doomed & F_SIGNAL)
+	{
+		ppm = PPM_UNDEFINED;
+		level = LEVEL_UNDEFINED;
+		setRepeat(0);
+		memset(country_code, 0, sizeof(country_code));
+		clearOpChannels();
+		msg.clear();
+	}
+}
+
 std::string getSprite(const Ship *ship)
 {
 	std::string shipofs = (ship->speed != SPEED_UNDEFINED && ship->speed > 0.5) ? "<y>88</y><w>20</w><h>20</h>" : "<y>68</y><w>20</w><h>20</h>";
@@ -168,7 +293,16 @@ int Ship::getMMSItype()
 	{
 		return MMSI_SARTEPIRB;
 	}
-	if (msg_type & ATON_MASK || (mmsi >= 990000000 && mmsi <= 999999999))
+	// the number outranks the message types: a bad decode must not reclassify a station
+	if (mmsi >= 990000000 && mmsi <= 999999999)
+	{
+		return MMSI_ATON;
+	}
+	if (mmsi < 9000000)
+	{
+		return MMSI_BASESTATION;
+	}
+	if (msg_type & ATON_MASK)
 	{
 		return MMSI_ATON;
 	}
@@ -180,7 +314,7 @@ int Ship::getMMSItype()
 	{
 		return MMSI_CLASS_B;
 	}
-	if (msg_type & BASESTATION_MASK || (mmsi < 9000000))
+	if (msg_type & BASESTATION_MASK)
 	{
 		return MMSI_BASESTATION;
 	}