commit 95d2e15e370d51c762a37eda4613acbef229ac4e
parent 47198d5d6b1fa5db80778b07e8cec69a6f87f588
Author: MTRNord <mtrnord1@gmail.com>
Date: Mon, 22 May 2023 12:48:31 +0200
Make sure that we write the commits properly and make sure to not load all changesets at once into ram since that will be ~200GB. Instead we load single files at a time
Diffstat:
6 files changed, 294 insertions(+), 188 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -751,6 +751,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
[[package]]
+name = "memmap2"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a0aa1b505aeecb0adb017db2b6a79a17a38e64f882a201f05e9de8a982cd6096"
+dependencies = [
+ "libc",
+]
+
+[[package]]
name = "mime"
version = "0.3.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -848,6 +857,7 @@ dependencies = [
"color-eyre",
"flate2",
"git2",
+ "memmap2",
"quick-xml",
"reqwest",
"serde",
diff --git a/Cargo.toml b/Cargo.toml
@@ -11,6 +11,7 @@ clap = { version = "4.3.0", features = ["derive"] }
color-eyre = "0.6.2"
flate2 = "1.0.26"
git2 = "0.17.1"
+memmap2 = "0.6.1"
quick-xml = { version = "0.28.2", features = ["async-tokio", "encoding", "escape-html", "overlapped-lists"] }
reqwest = { version = "0.11.18", default-features = false, features = ["rustls-tls", "gzip", "stream", "trust-dns"] }
serde = { version = "1.0.163", features = ["derive"] }
diff --git a/src/git/mod.rs b/src/git/mod.rs
@@ -107,6 +107,7 @@ pub fn commit(
for file in removed_files {
index.remove_path(std::path::Path::new(&file))?;
}
+ index.write()?;
index.write_tree()?
};
let tree = repository.find_tree(tree_id)?;
diff --git a/src/main.rs b/src/main.rs
@@ -1,14 +1,12 @@
-use std::time::Duration;
+use std::{fs::File, time::Duration};
use clap::Parser;
use color_eyre::eyre::Result;
use git2::Signature;
+use memmap2::Mmap;
use tracing::{info, warn};
-use crate::{
- git::init_git_repository,
- osm::{changesets::parse_changeset, osm_data::convert_objects_to_git},
-};
+use crate::{git::init_git_repository, osm::osm_data::convert_objects_to_git};
mod git;
mod osm;
@@ -88,7 +86,6 @@ async fn main() -> Result<()> {
let mut changeset_position_middle_incremented = false;
let mut changeset_position_top_incremented = false;
- let mut changesets = Vec::new();
loop {
// Check for cache and use it if it exists
let cache_file_path = format!(
@@ -100,68 +97,74 @@ async fn main() -> Result<()> {
);
if std::path::Path::new(&cache_file_path).exists() {
- info!("Using cached changeset file at {}", cache_file_path);
- let changeset_data = std::fs::read(&cache_file_path)?;
- let parsed_changeset = parse_changeset(changeset_data.into())?;
- info!("Changeset file parsed");
- changesets.extend(parsed_changeset);
+ info!(
+ "We already got the changeset file at {}. Skipping",
+ cache_file_path
+ );
// Increment the changeset position
changeset_position_bottom += 1;
changeset_position_middle_incremented = false;
changeset_position_top_incremented = false;
} else {
- // First we download the changeset files
- let changeset_url = format!(
- "{}/{:03}/{:03}/{:03}.osm.gz",
- cli.changeset_server,
- changeset_position_top,
- changeset_position_middle,
- changeset_position_bottom
- );
- info!("Downloading changeset file from {}", changeset_url);
- let changeset_response: reqwest::Response = client.get(&changeset_url).send().await?;
- if changeset_response.status() == reqwest::StatusCode::NOT_FOUND {
- warn!("Changeset file not found at {}", changeset_url);
- // We've reached the end of the changesets for this bottom position.
- // If we incremented top and failed again, we're done.
- if changeset_position_top_incremented {
- info!("Finished or failed downloading changesets");
- info!(
- "Changeset position: {} {} {}",
- changeset_position_top,
- changeset_position_middle,
- changeset_position_bottom
- );
- warn!("Response body: {:?}", changeset_response.text().await?);
- // TODO: We want to have an endless loop here optionally so that we can keep trying to download changesets.
- break;
+ {
+ // First we download the changeset files
+ let changeset_url = format!(
+ "{}/{:03}/{:03}/{:03}.osm.gz",
+ cli.changeset_server,
+ changeset_position_top,
+ changeset_position_middle,
+ changeset_position_bottom
+ );
+ info!("Downloading changeset file from {}", changeset_url);
+ let changeset_response: reqwest::Response =
+ client.get(&changeset_url).send().await?;
+ if changeset_response.status() == reqwest::StatusCode::NOT_FOUND {
+ warn!("Changeset file not found at {}", changeset_url);
+ // We've reached the end of the changesets for this bottom position.
+ // If we incremented top and failed again, we're done.
+ if changeset_position_top_incremented {
+ info!("Finished or failed downloading changesets");
+ info!(
+ "Changeset position: {} {} {}",
+ changeset_position_top,
+ changeset_position_middle,
+ changeset_position_bottom
+ );
+ warn!("Response body: {:?}", changeset_response.text().await?);
+ // TODO: We want to have an endless loop here optionally so that we can keep trying to download changesets.
+ break;
+ }
+ // We reset bottom to 0 and increment middle.
+ // We also mark middle as incremented so that we increment top on the next failure.
+ if !changeset_position_middle_incremented && changeset_position_bottom != 0 {
+ changeset_position_bottom = 0;
+ changeset_position_middle += 1;
+ changeset_position_middle_incremented = true;
+ changeset_position_top_incremented = false;
+ } else {
+ changeset_position_middle_incremented = false;
+ changeset_position_top += 1;
+ changeset_position_top_incremented = true;
+ changeset_position_middle = 0;
+ changeset_position_bottom = 0;
+ }
+ continue;
}
- // We reset bottom to 0 and increment middle.
- // We also mark middle as incremented so that we increment top on the next failure.
- if !changeset_position_middle_incremented && changeset_position_bottom != 0 {
- changeset_position_bottom = 0;
- changeset_position_middle += 1;
- changeset_position_middle_incremented = true;
- changeset_position_top_incremented = false;
- } else {
- changeset_position_middle_incremented = false;
- changeset_position_top += 1;
- changeset_position_top_incremented = true;
- changeset_position_middle = 0;
- changeset_position_bottom = 0;
- }
- continue;
- }
- let changeset_data = changeset_response.bytes().await?;
- info!("Caching changeset file to disk");
- std::fs::create_dir_all(std::path::Path::new(&cache_file_path).parent().unwrap())?;
- std::fs::write(&cache_file_path, &changeset_data)?;
- info!("Changeset file downloaded");
+ let changeset_data = changeset_response.bytes().await?;
+ info!("Caching changeset file to disk");
+ std::fs::create_dir_all(std::path::Path::new(&cache_file_path).parent().unwrap())?;
+ std::fs::write(&cache_file_path, &changeset_data)?;
+ info!("Changeset file downloaded");
+ };
+
+ // TODO: We need to dynamically do this based on the data instead. Otherwise we dont have ram larrge enough
+ // let file = File::open(cache_file_path)?;
+ // let changeset_data = unsafe { Mmap::map(&file)? };
- let parsed_changeset = parse_changeset(changeset_data)?;
- info!("Changeset file parsed");
- changesets.extend(parsed_changeset);
+ // let parsed_changeset = parse_changeset(&changeset_data)?;
+ // info!("Changeset file parsed");
+ // changesets.extend(parsed_changeset);
// Increment the changeset position
changeset_position_bottom += 1;
@@ -193,8 +196,9 @@ async fn main() -> Result<()> {
if std::path::Path::new(&cache_file_path).exists() {
info!("Using cached data file at {}", cache_file_path);
- let data = std::fs::read(&cache_file_path)?;
- convert_objects_to_git(&repository, &author, &changesets, &data)?;
+ let file = File::open(&cache_file_path)?;
+ let data = unsafe { Mmap::map(&file)? };
+ convert_objects_to_git(&repository, &author, &data, cli.cache_path.clone())?;
info!("Data file parsed");
// Increment the data position
@@ -202,55 +206,61 @@ async fn main() -> Result<()> {
data_position_middle_incremented = false;
data_position_top_incremented = false;
} else {
- // Download minute replication files and find the changesets that were modified in that minute
- let data_url = format!(
- "{}/{:03}/{:03}/{:03}.osc.gz",
- cli.replication_server,
- data_position_top,
- data_position_middle,
- data_position_bottom
- );
- info!("Downloading data file from {}", data_url);
- let data_response: reqwest::Response = client.get(&data_url).send().await?;
+ {
+ // Download minute replication files and find the changesets that were modified in that minute
+ let data_url = format!(
+ "{}/{:03}/{:03}/{:03}.osc.gz",
+ cli.replication_server,
+ data_position_top,
+ data_position_middle,
+ data_position_bottom
+ );
+ info!("Downloading data file from {}", data_url);
+ let data_response: reqwest::Response = client.get(&data_url).send().await?;
- if data_response.status() == reqwest::StatusCode::NOT_FOUND {
- warn!("data file not found at {}", data_url);
- // We've reached the end of the data for this bottom position.
- // If we incremented top and failed again, we're done.
- if data_position_top_incremented {
- info!("Finished or failed downloading data");
- info!(
- "Data position: {} {} {}",
- data_position_top, data_position_middle, data_position_bottom
- );
- warn!("Response body: {:?}", data_response.text().await?);
+ if data_response.status() == reqwest::StatusCode::NOT_FOUND {
+ warn!("data file not found at {}", data_url);
+ // We've reached the end of the data for this bottom position.
+ // If we incremented top and failed again, we're done.
+ if data_position_top_incremented {
+ info!("Finished or failed downloading data");
+ info!(
+ "Data position: {} {} {}",
+ data_position_top, data_position_middle, data_position_bottom
+ );
+ warn!("Response body: {:?}", data_response.text().await?);
- // TODO: We want to have an endless loop here optionally so that we can keep trying to download changesets.
- break;
- }
- // We reset bottom to 0 and increment middle.
- // We also mark middle as incremented so that we increment top on the next failure.
- if !data_position_middle_incremented && data_position_bottom != 0 {
- data_position_bottom = 0;
- data_position_middle += 1;
- data_position_middle_incremented = true;
- data_position_top_incremented = false;
- } else {
- data_position_middle_incremented = false;
- data_position_top += 1;
- data_position_top_incremented = true;
- data_position_middle = 0;
- data_position_bottom = 0;
+ // TODO: We want to have an endless loop here optionally so that we can keep trying to download changesets.
+ break;
+ }
+ // We reset bottom to 0 and increment middle.
+ // We also mark middle as incremented so that we increment top on the next failure.
+ if !data_position_middle_incremented && data_position_bottom != 0 {
+ data_position_bottom = 0;
+ data_position_middle += 1;
+ data_position_middle_incremented = true;
+ data_position_top_incremented = false;
+ } else {
+ data_position_middle_incremented = false;
+ data_position_top += 1;
+ data_position_top_incremented = true;
+ data_position_middle = 0;
+ data_position_bottom = 0;
+ }
+ continue;
}
- continue;
- }
- let data = data_response.bytes().await?;
- info!("Caching Data file to disk");
- std::fs::create_dir_all(std::path::Path::new(&cache_file_path).parent().unwrap())?;
- std::fs::write(&cache_file_path, &data)?;
- info!("Data file downloaded");
- convert_objects_to_git(&repository, &author, &changesets, &data)?;
+ let data = data_response.bytes().await?;
+ info!("Caching Data file to disk");
+ std::fs::create_dir_all(std::path::Path::new(&cache_file_path).parent().unwrap())?;
+ std::fs::write(&cache_file_path, &data)?;
+ info!("Data file downloaded");
+ };
+
+ let file = File::open(cache_file_path)?;
+ let data = unsafe { Mmap::map(&file)? };
+
+ convert_objects_to_git(&repository, &author, &data, cli.cache_path.clone())?;
// Increment the data position
data_position_bottom += 1;
diff --git a/src/osm/changesets.rs b/src/osm/changesets.rs
@@ -1,4 +1,3 @@
-use bytes::Bytes;
use color_eyre::eyre::Result;
use flate2::bufread::GzDecoder;
use quick_xml::{
@@ -137,13 +136,13 @@ impl Changeset {
}
}
-pub fn parse_changeset(changeset_data: Bytes) -> Result<Vec<Changeset>> {
+pub fn parse_changeset(changeset_data: &[u8]) -> Result<Vec<Changeset>> {
// If file is empty, return an empty vector
if changeset_data.is_empty() {
return Ok(Vec::new());
}
// Decompress the changeset file
- let mut changeset_data_reader = GzDecoder::new(&changeset_data[..]);
+ let mut changeset_data_reader = GzDecoder::new(changeset_data);
let mut changeset_data = String::new();
changeset_data_reader.read_to_string(&mut changeset_data)?;
debug!(
diff --git a/src/osm/osm_data.rs b/src/osm/osm_data.rs
@@ -11,14 +11,16 @@ use std::{
borrow::Cow,
collections::BTreeMap,
convert::Infallible,
+ fs::File,
io::{Read, Write},
+ path::Path,
};
use time::{format_description::well_known::Iso8601, OffsetDateTime};
use tracing::{debug, error, info, warn};
use crate::git::commit;
-use super::changesets::Changeset;
+use super::changesets::{parse_changeset, Changeset};
const FILE_VERSION: &str = "0.1.0";
@@ -468,8 +470,8 @@ pub enum OSMObject {
pub fn convert_objects_to_git(
repository: &Repository,
committer: &Signature,
- changesets: &[Changeset],
data: &[u8],
+ cache_folder: String,
) -> Result<()> {
// If the file is empty we skip it
if data.is_empty() {
@@ -845,83 +847,166 @@ pub fn convert_objects_to_git(
.collect();
for changeset in changeset_list {
- // Construct the commit and apply it
- let changeset = changesets
- .iter()
- .find(|c| c.id == *changeset)
- .expect("Unable to find changeset");
-
- // Get comment tag if it exists and trim it
- let comment = changeset
- .tags
- .get("comment")
- .map(|s| s.trim())
- .unwrap_or("");
-
- // Parse changeset time (ISO 8601) to git time (seconds since epoch) with offset 0 (UTC) using `time`
- let changeset_time = changeset
- .closed_at
- .clone()
- .unwrap_or(changeset.created_at.clone());
- let commit_time =
- OffsetDateTime::parse(changeset_time.as_str(), &Iso8601::DEFAULT)?.unix_timestamp();
-
- let author = git2::Signature::new(
- &changeset.user,
- &format!("{}@osm", changeset.user),
- &Time::new(commit_time, 0),
- )
- .expect("Unable to create author signature");
-
- let repository_folder = repository.path().parent().unwrap();
-
- let added_or_changed_files = created_or_modified_objects_for_changeset
- .get(&changeset.id)
- .unwrap_or(&Vec::new())
- .iter()
- .map(|object| match object {
- OSMObject::Node(ref node) => repository_folder.join(format!("{}.yaml", node.id)),
- OSMObject::Way(ref way) => repository_folder.join(format!("{}.yaml", way.id)),
- OSMObject::Relation(ref relation) => {
- repository_folder.join(format!("{}.yaml", relation.id))
- }
- })
- .map(|path| path.to_string_lossy().to_string())
- .collect::<Vec<String>>();
-
- let removed_files = deleted_objects_for_changeset
- .get(&changeset.id)
- .unwrap_or(&Vec::new())
- .iter()
- .map(|object| match object {
- OSMObject::Node(ref node) => repository_folder.join(format!("{}.yaml", node.id)),
- OSMObject::Way(ref way) => repository_folder.join(format!("{}.yaml", way.id)),
- OSMObject::Relation(ref relation) => {
- repository_folder.join(format!("{}.yaml", relation.id))
- }
- })
- .map(|path| path.to_string_lossy().to_string())
- .collect::<Vec<String>>();
-
- let oid = commit(
- repository,
- added_or_changed_files,
- removed_files,
- comment,
- &author,
- committer,
- )?;
-
- // Convert tags to "Key: Value" strings separated by newlines for the note
- let note = changeset
- .tags
- .iter()
- .map(|(key, value)| format!("{}: {}", key, value))
- .collect::<Vec<String>>()
- .join("\n");
-
- repository.note(&author, committer, None, oid, ¬e, false)?;
+ // Find the changeset within the files of the cache
+ let changeset =
+ find_changesets_in_cache(cache_folder.clone(), *changeset, None, None, None)?;
+
+ if let Some(changeset) = changeset {
+ // Get comment tag if it exists and trim it
+ let comment = changeset
+ .tags
+ .get("comment")
+ .map(|s| s.trim())
+ .unwrap_or("");
+
+ // Parse changeset time (ISO 8601) to git time (seconds since epoch) with offset 0 (UTC) using `time`
+ let changeset_time = changeset
+ .closed_at
+ .clone()
+ .unwrap_or(changeset.created_at.clone());
+ let commit_time =
+ OffsetDateTime::parse(changeset_time.as_str(), &Iso8601::DEFAULT)?.unix_timestamp();
+
+ let author = git2::Signature::new(
+ &changeset.user,
+ &format!("{}@osm", changeset.user),
+ &Time::new(commit_time, 0),
+ )
+ .expect("Unable to create author signature");
+
+ let repository_folder = repository.path().parent().unwrap();
+
+ let added_or_changed_files = created_or_modified_objects_for_changeset
+ .get(&changeset.id)
+ .unwrap_or(&Vec::new())
+ .iter()
+ .map(|object| match object {
+ OSMObject::Node(ref node) => {
+ repository_folder.join(format!("{}.yaml", node.id))
+ }
+ OSMObject::Way(ref way) => repository_folder.join(format!("{}.yaml", way.id)),
+ OSMObject::Relation(ref relation) => {
+ repository_folder.join(format!("{}.yaml", relation.id))
+ }
+ })
+ .map(|path| path.to_string_lossy().to_string())
+ .collect::<Vec<String>>();
+
+ let removed_files = deleted_objects_for_changeset
+ .get(&changeset.id)
+ .unwrap_or(&Vec::new())
+ .iter()
+ .map(|object| match object {
+ OSMObject::Node(ref node) => {
+ repository_folder.join(format!("{}.yaml", node.id))
+ }
+ OSMObject::Way(ref way) => repository_folder.join(format!("{}.yaml", way.id)),
+ OSMObject::Relation(ref relation) => {
+ repository_folder.join(format!("{}.yaml", relation.id))
+ }
+ })
+ .map(|path| path.to_string_lossy().to_string())
+ .collect::<Vec<String>>();
+
+ let oid = commit(
+ repository,
+ added_or_changed_files,
+ removed_files,
+ comment,
+ &author,
+ committer,
+ )?;
+
+ // Convert tags to "Key: Value" strings separated by newlines for the note
+ let note = changeset
+ .tags
+ .iter()
+ .map(|(key, value)| format!("{}: {}", key, value))
+ .collect::<Vec<String>>()
+ .join("\n");
+
+ repository.note(&author, committer, None, oid, ¬e, false)?;
+
+ // We make sure to not keep it around since this would get large quite quickly
+ drop(changeset);
+ }
}
Ok(())
}
+
+/// Scans the files in the cache folder and returns the requested changeset
+///
+/// # Arguments
+///
+/// * `cache_folder` - The folder where the changesets are stored
+/// * `changeset_id` - The id of the changeset to find
+///
+/// # Returns
+///
+/// The changeset if found
+fn find_changesets_in_cache(
+ cache_folder: String,
+ changeset_id: u64,
+ changeset_position_top: Option<u64>,
+ changeset_position_middle: Option<u64>,
+ changeset_position_bottom: Option<u64>,
+) -> Result<Option<Changeset>> {
+ // Changesets are of the format "{cache_folder}/changesets/{:03}/{:03}/{:03}.osm.gz".
+ //
+ // Each file has to be parsed using "parse_changeset(&changeset_data)?;"
+
+ let changeset_folder = format!("{}/changesets", cache_folder);
+ let mut changeset_position_top = changeset_position_top.unwrap_or(0);
+ let mut changeset_position_middle = changeset_position_middle.unwrap_or(0);
+ let mut changeset_position_bottom = changeset_position_bottom.unwrap_or(0);
+ let changeset_file = format!(
+ "{:03}/{:03}/{:03}.osm.gz",
+ changeset_position_top, changeset_position_middle, changeset_position_bottom
+ );
+ let changeset_path = format!("{}/{}", changeset_folder, changeset_file);
+
+ if !Path::new(&changeset_path).exists() {
+ return Ok(None);
+ }
+
+ let mut changeset_file = File::open(changeset_path)?;
+ let mut changeset_data = Vec::new();
+ changeset_file.read_to_end(&mut changeset_data)?;
+
+ let changesets = parse_changeset(&changeset_data)?;
+
+ // Check if the file has the correct changeset id in vector otherwise we recurse to the next file
+ if changesets.iter().any(|c| c.id == changeset_id) {
+ return Ok(changesets.into_iter().find(|c| c.id == changeset_id));
+ }
+
+ // We recurse to the next file since we found no changeset with the correct id
+
+ if changeset_position_top == 999
+ && changeset_position_middle == 999
+ && changeset_position_bottom == 999
+ {
+ // Uhhhhhh?!
+ return Ok(None);
+ }
+
+ if changeset_position_middle == 999 && changeset_position_bottom == 999 {
+ changeset_position_middle = 0;
+ changeset_position_bottom = 0;
+ changeset_position_top += 1;
+ }
+
+ if changeset_position_bottom == 999 {
+ changeset_position_bottom = 0;
+ changeset_position_middle += 1;
+ }
+
+ find_changesets_in_cache(
+ cache_folder,
+ changeset_id,
+ Some(changeset_position_top),
+ Some(changeset_position_middle),
+ Some(changeset_position_bottom),
+ )
+}
