diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 000000000..5cb3e65ba --- /dev/null +++ b/.gitmodules @@ -0,0 +1,4 @@ +[submodule "examples/nexmark"] + path = examples/nexmark + url = https://github.com/nexmark/nexmark/ + branch = master diff --git a/benchmark/Puppetfile b/benchmark/Puppetfile new file mode 100644 index 000000000..e71bdecc9 --- /dev/null +++ b/benchmark/Puppetfile @@ -0,0 +1,11 @@ +mod 'my', local: true +mod 'geoffwilliams-filemagic', '0.6.1' +mod 'puppet-python', '3.0.1' +mod 'puppet-squid', '2.2.2' +mod 'puppet-alternatives', '3.0.0' +mod 'puppetlabs-apt', '7.6.0' +mod 'puppetlabs-concat', '6.1.0' +mod 'puppetlabs-sshkeys_core' +mod 'puppetlabs-stdlib', '6.0.0' +mod 'puppetlabs-vcsrepo', '3.0.0' +mod 'saz-ssh', '6.2.0' diff --git a/benchmark/ansible/flamestream.yml b/benchmark/ansible/flamestream.yml index 60e4f8350..9b3c4a118 100644 --- a/benchmark/ansible/flamestream.yml +++ b/benchmark/ansible/flamestream.yml @@ -1,28 +1,19 @@ --- -- name: Install Python - hosts: all - gather_facts: false - tasks: - - name: Install Python 2.x - raw: test -e /usr/bin/python || (apt update && apt install -y python-simplejson) - register: test - changed_when: test.stdout - -- name: Prepare cluster +- name: Deliver flamestream artifacts hosts: all roles: - - { role: common-packages } + - { role: flamestream-common } + +- name: Stop workers + hosts: workers + tasks: + - shell: "cd {{ flamestream_dir }} && bash entrypoint.sh stop" - name: Setup zookeeper hosts: manager roles: - { role: zookeeper } -- name: Deliver flamestream artifacts - hosts: all - roles: - - { role: flamestream-common } - - hosts: workers tasks: - file: @@ -38,33 +29,3 @@ tasks: - wait_for: path: /tmp/flame_stream - -#- name: Pause -# hosts: bench -# tasks: -# - pause: - -- name: Run the benchmark - hosts: bench - roles: - - { role: flamestream-bench , bench_class: com.spbsu.flamestream.example.benchmark.BenchStand } - -- name: Destroy flamestream - hosts: workers - roles: - - { role: flamestream-destroy } - -- name: Destroy zookeeper - hosts: manager - roles: - - { role: zookeeper-destroy } - -- hosts: all - tasks: - - name: Fetch traces - fetch: - src: /tmp/trace.csv - dest: "{{ results_dir }}/flamestream/traces/{{ inventory_hostname }}.csv" - flat: true - tags: - - lat diff --git a/benchmark/ansible/group_vars/all.yml b/benchmark/ansible/group_vars/all.yml index 2fce1d951..52f01d4cd 100644 --- a/benchmark/ansible/group_vars/all.yml +++ b/benchmark/ansible/group_vars/all.yml @@ -1,18 +1,18 @@ -rate: 50 +local_acker_flush_delay_in_millis: 10 guarantees: AT_MOST_ONCE -millis_between_commits: 50 +millis_between_commits: 0 validator: Wiki1000Validator -base_dir: /opt +base_dir: "{{ ansible_env.HOME }}" flamestream_dir: "{{ base_dir }}/flamestream" zookeeper_dir: "{{ base_dir }}/zookeeper" bench_dir: "{{ base_dir }}/flamestream" flink_dir: "{{ base_dir }}/flink" -tarball_dir: /opt/src +tarball_dir: "{{ base_dir }}/tarballs" results_dir: results/{{ lookup("pipe", "git rev-parse --short HEAD") }}/{{ groups['workers'] | length }}/{{ rate }}/{{ guarantees }}/{{ millis_between_commits }} -zookeeper_version: 3.5.8 +zookeeper_version: 3.6.2 zookeeper_client_port: 2181 flamestream_worker_port: 5412 diff --git a/benchmark/ansible/roles/flamestream-bench/tasks/main.yml b/benchmark/ansible/roles/flamestream-bench/tasks/main.yml index 3d80ba34c..3335cf71c 100644 --- a/benchmark/ansible/roles/flamestream-bench/tasks/main.yml +++ b/benchmark/ansible/roles/flamestream-bench/tasks/main.yml @@ -17,7 +17,7 @@ - name: Start the flamestream bench shell: > java - -Daeron.term.buffer.length=4194304 -Daeron.mtu.length=16384 -Xms500m -Xmx500m -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+HeapDumpOnOutOfMemoryError + -Daeron.term.buffer.length=4194304 -Daeron.mtu.length=16384 -Xms500m -Xmx1500m -Xlog:gc,gc+cpu=info::utc -XX:+HeapDumpOnOutOfMemoryError -cp '{{ flamestream_dir }}/lib/*' {{ bench_class }} {{ flamestream_dir }}/bench.conf {{ flamestream_dir }}/deployer.conf > {{ flamestream_dir }}/bench.log 2>&1 diff --git a/benchmark/ansible/roles/flamestream-bench/templates/bench.conf.j2 b/benchmark/ansible/roles/flamestream-bench/templates/bench.conf.j2 index 3a7fec4dd..38c4c9d71 100644 --- a/benchmark/ansible/roles/flamestream-bench/templates/bench.conf.j2 +++ b/benchmark/ansible/roles/flamestream-bench/templates/bench.conf.j2 @@ -1,11 +1,11 @@ benchmark { bench-host = "{{ groups['bench'][0] }}" - input-host = "{{ groups['input'][0] }}" + input-host = "{{ groups['workers'][0] }}" bench-source-port = 4567 bench-sink-port = 5678 wiki-dump-path = "{{ flamestream_dir }}/wiki.xml" sleep-between-docs-ms = {{ rate }} - worker-id = "{{ groups['worker'][0] }}" + worker-id = "{{ groups['workers'][0] }}" validator = com.spbsu.flamestream.example.benchmark.validators.{{ validator }} } diff --git a/benchmark/ansible/roles/flamestream-worker/tasks/main.yml b/benchmark/ansible/roles/flamestream-worker/tasks/main.yml index cbdca9ca2..6abb805b0 100644 --- a/benchmark/ansible/roles/flamestream-worker/tasks/main.yml +++ b/benchmark/ansible/roles/flamestream-worker/tasks/main.yml @@ -18,3 +18,5 @@ DEFAULT_MINIMAL_TIME: 0 MAX_ELEMENTS_IN_GRAPH: 100 MILLIS_BETWEEN_COMMITS: "{{ millis_between_commits }}" + ACKERS_NUMBER: 1 + ACKER_WINDOW: 10 diff --git a/benchmark/ansible/roles/zookeeper/tasks/main.yml b/benchmark/ansible/roles/zookeeper/tasks/main.yml index bdd0d1917..6072ed842 100644 --- a/benchmark/ansible/roles/zookeeper/tasks/main.yml +++ b/benchmark/ansible/roles/zookeeper/tasks/main.yml @@ -25,11 +25,18 @@ src: zoo.cfg.j2 dest: "{{ zookeeper_dir }}/conf/zoo.cfg" -- name: Restart zk - shell: "{{ zookeeper_dir }}/bin/zkServer.sh restart" +- name: Stop zk + shell: "{{ zookeeper_dir }}/bin/zkServer.sh stop" -- name: Clean graphs - shell: "{{ zookeeper_cli }} rmr /graph" +- file: + state: absent + path: "{{ zookeeper_dir }}/data/" + +- name: Start zk + shell: "{{ zookeeper_dir }}/bin/zkServer.sh start" + +- command: "{{ zookeeper_dir }}/bin/zkServer.sh status" + retries: 5 + register: result + until: result is not failed -- name: Clean configs - shell: "{{ zookeeper_cli }} rmr /config" diff --git a/benchmark/ansible/roles/zookeeper/vars/main.yml b/benchmark/ansible/roles/zookeeper/vars/main.yml index fa8b8bb43..79163017d 100644 --- a/benchmark/ansible/roles/zookeeper/vars/main.yml +++ b/benchmark/ansible/roles/zookeeper/vars/main.yml @@ -1,6 +1,6 @@ --- zookeeper_dir: "{{ base_dir }}/zookeeper" -zookeeper_url: http://www.apache.org/dist/zookeeper/zookeeper-{{ zookeeper_version }}/zookeeper-{{ zookeeper_version }}.tar.gz +zookeeper_url: http://www.apache.org/dist/zookeeper/zookeeper-{{ zookeeper_version }}/apache-zookeeper-{{ zookeeper_version }}-bin.tar.gz tick_time: 3000 zookeeper_data_dir: "{{ zookeeper_dir }}/data" -zookeeper_cli: "{{ zookeeper_dir }}/bin/zkCli.sh -server localhost:{{ zookeeper_client_port }}" \ No newline at end of file +zookeeper_cli: "{{ zookeeper_dir }}/bin/zkCli.sh -server localhost:{{ zookeeper_client_port }}" diff --git a/benchmark/aws/remote.tf b/benchmark/aws/remote.tf new file mode 100644 index 000000000..f595700a8 --- /dev/null +++ b/benchmark/aws/remote.tf @@ -0,0 +1,102 @@ +variable "access_key" {} +variable "secret_key" {} +variable "key_pair" {} +variable "cluster_size" { + default = 11 +} +variable "region" { + default = "us-east-2" +} + +provider "aws" { + access_key = "${var.access_key}" + secret_key = "${var.secret_key}" + region = "${var.region}" +} + +resource "aws_key_pair" "mega-benchmarker" { + key_name = "mega-benchmarker" + public_key = "${var.key_pair}" +} + +resource "aws_vpc" "workers" { + cidr_block = "10.0.0.0/24" +} + +resource "aws_internet_gateway" "workers" { + vpc_id = "${aws_vpc.workers.id}" +} + +resource "aws_subnet" "workers" { + vpc_id = "${aws_vpc.workers.id}" + cidr_block = "10.0.0.0/24" + map_public_ip_on_launch = true +} + +resource "aws_route_table" "workers" { + vpc_id = "${aws_vpc.workers.id}" + + route { + cidr_block = "0.0.0.0/0" + gateway_id = "${aws_internet_gateway.workers.id}" + } +} + +resource "aws_route_table_association" "public" { + subnet_id = "${aws_subnet.workers.id}" + route_table_id = "${aws_route_table.workers.id}" +} + +resource "aws_security_group" "workers" { + vpc_id = "${aws_vpc.workers.id}" + + ingress { + from_port = 0 + to_port = 0 + protocol = "-1" + cidr_blocks = [ + "${aws_vpc.workers.cidr_block}"] + } + + ingress { + from_port = 8081 + to_port = 8081 + protocol = "tcp" + cidr_blocks = [ + "0.0.0.0/0"] + } + + ingress { + from_port = 22 + to_port = 22 + protocol = "tcp" + cidr_blocks = [ + "0.0.0.0/0"] + } + + egress { + protocol = -1 + from_port = 0 + to_port = 0 + cidr_blocks = [ + "0.0.0.0/0"] + } +} + +resource "aws_instance" "worker" { + count = "${var.cluster_size}" + ami = "ami-965e6bf3" + instance_type = "t2.small" + key_name = "${aws_key_pair.mega-benchmarker.key_name}" + subnet_id = "${aws_subnet.workers.id}" + vpc_security_group_ids = [ + "${aws_security_group.workers.id}"] + associate_public_ip_address = true +} + +output "public_ips" { + value = "${join(", ", aws_instance.worker.*.public_ip)}" +} +output "private_ips" { + value = "${join(", ", aws_instance.worker.*.private_ip)}" +} diff --git a/benchmark/azure/remote.tf b/benchmark/azure/remote.tf new file mode 100644 index 000000000..ef1fc9d1e --- /dev/null +++ b/benchmark/azure/remote.tf @@ -0,0 +1,160 @@ +variable "access_key" {} +variable "secret_key" {} +variable "key_pair" {} +variable "cluster_size" { + default = 3 +} + +variable "prefix" { + default = "flamestream-benchmarks" +} + +provider "azurerm" { + version = "=1.33.0" +} + +data "azurerm_subscription" "current" {} + +resource "azurerm_resource_group" "main" { + name = "${var.prefix}-resources" + location = "West Europe" +} + +resource "azurerm_virtual_network" "main" { + name = "${var.prefix}-network" + address_space = ["10.0.0.0/16"] + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" +} + +resource "azurerm_subnet" "internal" { + name = "${var.prefix}-internal" + resource_group_name = "${azurerm_resource_group.main.name}" + virtual_network_name = "${azurerm_virtual_network.main.name}" + address_prefix = "10.0.2.0/24" +} + +resource "azurerm_public_ip" "manager" { + name = "test-pip" + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" + allocation_method = "Dynamic" + idle_timeout_in_minutes = 30 +} + +resource "azurerm_network_interface" "manager" { + name = "${var.prefix}-manager" + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" + + ip_configuration { + name = "testconfiguration1" + subnet_id = "${azurerm_subnet.internal.id}" + private_ip_address_allocation = "Dynamic" + public_ip_address_id = "${azurerm_public_ip.manager.id}" + } +} + +resource "azurerm_managed_disk" "manager" { + name = "${var.prefix}-manager" + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" + storage_account_type = "Standard_LRS" + create_option = "Copy" + source_resource_id = "${data.azurerm_subscription.current.id}/resourceGroups/FlamestreamBenchmarks/providers/Microsoft.Compute/snapshots/FlamestreamBenchmarks" + disk_size_gb = "30" +} + +resource "azurerm_virtual_machine" "manager" { + name = "${var.prefix}-manager" + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" + network_interface_ids = ["${azurerm_network_interface.manager.id}"] + vm_size = "Standard_B1ms" + + delete_os_disk_on_termination = true + delete_data_disks_on_termination = true + + storage_os_disk { + name = "${azurerm_managed_disk.manager.name}" + caching = "ReadWrite" + create_option = "Attach" + managed_disk_id = "${azurerm_managed_disk.manager.id}" + os_type = "Linux" + } + os_profile_linux_config { + disable_password_authentication = true + ssh_keys { + key_data = file("~/.ssh/id_rsa.pub") + path = "/home/ubuntu/.ssh/authorized_keys" + } + } +} + +data "azurerm_public_ip" "manager" { + name = "${azurerm_public_ip.manager.name}" + resource_group_name = "${azurerm_virtual_machine.manager.resource_group_name}" +} + +resource "azurerm_network_interface" "workers" { + count = var.cluster_size + name = "${var.prefix}-worker-${count.index}" + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" + + ip_configuration { + name = "testconfiguration1" + subnet_id = "${azurerm_subnet.internal.id}" + private_ip_address_allocation = "Dynamic" + } +} + +resource "azurerm_managed_disk" "workers" { + count = var.cluster_size + name = "${var.prefix}-worker-${count.index}" + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" + storage_account_type = "Standard_LRS" + create_option = "Copy" + source_resource_id = "${data.azurerm_subscription.current.id}/resourceGroups/FlamestreamBenchmarks/providers/Microsoft.Compute/snapshots/FlamestreamBenchmarks" + disk_size_gb = "30" +} + +resource "azurerm_virtual_machine" "workers" { + count = var.cluster_size + name = "${var.prefix}-worker-${count.index}" + location = "${azurerm_resource_group.main.location}" + resource_group_name = "${azurerm_resource_group.main.name}" + network_interface_ids = ["${azurerm_network_interface.workers[count.index].id}"] + vm_size = "Standard_B1ms" + + delete_os_disk_on_termination = true + delete_data_disks_on_termination = true + + storage_os_disk { + name = azurerm_managed_disk.workers[count.index].name + caching = "ReadWrite" + create_option = "Attach" + managed_disk_id = azurerm_managed_disk.workers[count.index].id + os_type = "Linux" + } + os_profile_linux_config { + disable_password_authentication = true + ssh_keys { + key_data = file("~/.ssh/id_rsa.pub") + path = "/home/ubuntu/.ssh/authorized_keys" + } + } +} + +output "manager_public_ip" { + value = "${data.azurerm_public_ip.manager.ip_address}" +} + +output "manager_private_ip" { + value = "${azurerm_network_interface.manager.private_ip_address}" +} + +output "worker_private_ips" { + value = "${azurerm_network_interface.workers.*.private_ip_address}" +} diff --git a/benchmark/bolt.yaml b/benchmark/bolt.yaml new file mode 100644 index 000000000..000032fc4 --- /dev/null +++ b/benchmark/bolt.yaml @@ -0,0 +1,3 @@ +modulepath: modules +ssh: + host-key-check: false diff --git a/benchmark/google/remote.tf b/benchmark/google/remote.tf new file mode 100644 index 000000000..941107884 --- /dev/null +++ b/benchmark/google/remote.tf @@ -0,0 +1,91 @@ +variable "google_project" {} +variable "cluster_size" {} + +provider "google" { + credentials = file("google_credentials.json") + project = var.google_project + region = "europe-north1" + zone = "europe-north1-c" +} + +resource "google_compute_firewall" "main" { + name = "flamestream" + network = "default" + source_ranges = [google_compute_subnetwork.main.ip_cidr_range] + + allow { + protocol = "tcp" + } + + allow { + protocol = "udp" + } +} + +resource "google_compute_subnetwork" "main" { + name = "flamestream" + ip_cidr_range = "10.0.0.0/24" + network = "default" +} + +resource "google_compute_disk" "manager" { + name = "flamestream-manager" + type = "pd-ssd" + zone = "europe-north1-c" + snapshot = "ekf3eap91l7v" +} + +resource "google_compute_instance" "manager" { + name = "flamestream-manager" + machine_type = "n1-standard-1" + + boot_disk { + source = google_compute_disk.manager.self_link + } + + metadata = { + sshKeys = "ubuntu:${file("~/.ssh/id_rsa.pub")}" + } + + network_interface { + subnetwork = google_compute_subnetwork.main.self_link + access_config { + } + } +} + +resource "google_compute_disk" "workers" { + count = var.cluster_size + name = "flamestream-worker-${count.index}" + type = "pd-ssd" + zone = "europe-north1-c" + snapshot = "ekf3eap91l7v" +} + +resource "google_compute_instance" "workers" { + count = var.cluster_size + name = "flamestream-worker-${count.index}" + machine_type = "n1-standard-1" + + boot_disk { + source = google_compute_disk.workers[count.index].self_link + } + + metadata = { + sshKeys = "ubuntu:${file("~/.ssh/id_rsa.pub")}" + } + + network_interface { + subnetwork = google_compute_subnetwork.main.self_link + } +} + +output "manager_public_ip" { + value = google_compute_instance.manager.network_interface.0.access_config.0.nat_ip +} +output "manager_private_ip" { + value = google_compute_instance.manager.network_interface.0.network_ip +} +output "worker_private_ips" { + value = google_compute_instance.workers.*.network_interface.0.network_ip +} diff --git a/benchmark/modules/my/lib/facter/etc-ssh-ssh_host_ecdsa_key.pub.rb b/benchmark/modules/my/lib/facter/etc-ssh-ssh_host_ecdsa_key.pub.rb new file mode 100644 index 000000000..e011adbd9 --- /dev/null +++ b/benchmark/modules/my/lib/facter/etc-ssh-ssh_host_ecdsa_key.pub.rb @@ -0,0 +1,6 @@ +Facter.add('/etc/ssh/ssh_host_ecdsa_key_pub') do + setcode do + file = '/etc/ssh/ssh_host_ecdsa_key.pub' + Facter::Util::Resolution.exec("[ -r #{file} ] && cat #{file}") + end +end diff --git a/benchmark/modules/my/lib/facter/home.rb b/benchmark/modules/my/lib/facter/home.rb new file mode 100644 index 000000000..96077dcbf --- /dev/null +++ b/benchmark/modules/my/lib/facter/home.rb @@ -0,0 +1,5 @@ +Facter.add('home') do + setcode do + Facter::Util::Resolution.exec('echo $HOME') + end +end diff --git a/benchmark/modules/my/lib/facter/ssh-id_rsa.pub.rb b/benchmark/modules/my/lib/facter/ssh-id_rsa.pub.rb new file mode 100644 index 000000000..7dc49f08e --- /dev/null +++ b/benchmark/modules/my/lib/facter/ssh-id_rsa.pub.rb @@ -0,0 +1,5 @@ +Facter.add('ssh_id_rsa_pub') do + setcode do + Facter::Util::Resolution.exec('cat ~/.ssh/id_rsa.pub') + end +end diff --git a/benchmark/modules/my/plans/bench.pp b/benchmark/modules/my/plans/bench.pp new file mode 100644 index 000000000..a83c9e18d --- /dev/null +++ b/benchmark/modules/my/plans/bench.pp @@ -0,0 +1,22 @@ +plan my::bench() { + get_targets('managers').apply_prep + $manager_host = get_target('managers').name + $worker_hosts = get_targets('workers').map |Target $worker| { $worker.name } + apply(get_targets('managers')) { + vcsrepo { "${facts['home']}/FlameStream": + ensure => present, + provider => git, + source => 'https://github.com/flame-stream/FlameStream', + revision => 'feature/labels-wip', + } + file { "${facts['home']}/FlameStream/benchmark/ansible/remote.yml": + content => inline_template(@(ERB)) +<%= { "all" => { "children" => { + "bench" => { "hosts" => { @manager_host => {} } }, + "manager" => { "hosts" => { @manager_host => {} } }, + "workers" => { "hosts" => @worker_hosts.map { |host| [host, {}] }.to_h }, +} } }.to_yaml %> +ERB + } + } +} diff --git a/benchmark/modules/my/plans/destroy.pp b/benchmark/modules/my/plans/destroy.pp new file mode 100644 index 000000000..f8d236378 --- /dev/null +++ b/benchmark/modules/my/plans/destroy.pp @@ -0,0 +1,29 @@ +plan my::destroy() { + $all = get_targets("all") + $manager = get_target("managers") + $all.apply_prep + 'localhost'.apply_prep + apply('localhost') { + $all.each |Target $target| { + $pub = $target.facts['/etc/ssh/ssh_host_ecdsa_key_pub'].split(' ') + sshkey { $target.uri: + ensure => absent, + type => $pub[0], + key => $pub[1], + host_aliases => if $target.vars['public_ip'] { $target.vars['public_ip'] } else { $target.vars['private_ip'] }, + target => "${facts['home']}/.ssh/known_hosts", + } + } + } + + apply('localhost') { + file_line { "~/.ssh/config:Include flamestream-benchmarks.config": + path => "${facts['home']}/.ssh/config", + line => "Include flamestream-benchmarks.config", + ensure => absent, + } + file { "${facts['home']}/.ssh/flamestream-benchmarks.config": + ensure => absent, + } + } +} diff --git a/benchmark/modules/my/plans/hosts.pp b/benchmark/modules/my/plans/hosts.pp new file mode 100644 index 000000000..ffe6d63a7 --- /dev/null +++ b/benchmark/modules/my/plans/hosts.pp @@ -0,0 +1,7 @@ +plan my::hosts() { + $all = get_targets('all') + $all.apply_prep + apply($all) { + $all.each |Target $target| { host { $target.name: ip => $target.vars['private_ip'] } } + } +} diff --git a/benchmark/modules/my/plans/http_proxy.pp b/benchmark/modules/my/plans/http_proxy.pp new file mode 100644 index 000000000..fa2b6eeb7 --- /dev/null +++ b/benchmark/modules/my/plans/http_proxy.pp @@ -0,0 +1,29 @@ +plan my::http_proxy() { + $manager = get_target('managers') + $manager.apply_prep + apply($manager) { + class { 'squid': + http_ports => { "${manager.vars['private_ip']}:3128" => { } }, + acls => { "all" => { + type => "src", + entries => ["0.0.0.0/0"], + }}, + http_access => { "all" => { + action => 'allow', + }}, + } + } + get_targets('workers').apply_prep + apply(get_targets('workers')) { + file_line { "http_proxy_env": + ensure => present, + line => "http_proxy=http://${manager.vars['private_ip']}:3128", + path => "/etc/environment", + } + file_line { "https_proxy_env": + ensure => present, + line => "https_proxy=http://${manager.vars['private_ip']}:3128", + path => "/etc/environment", + } + } +} diff --git a/benchmark/modules/my/plans/install.pp b/benchmark/modules/my/plans/install.pp new file mode 100644 index 000000000..92412255c --- /dev/null +++ b/benchmark/modules/my/plans/install.pp @@ -0,0 +1,32 @@ +plan my::install() { + $manager = get_targets('managers') + $manager.apply_prep + apply($manager) { + package { 'git': } + package { 'libffi-dev': } + package { 'libzmq3-dev': } + package { 'maven': } + package { 'python3': } + package { 'python3.9': } + package { 'python3.9-dev': } + package { 'python3.9-distutils': } + class { 'python::pip::bootstrap': version => 'pip3' } + python::pip { 'ansible': } + } + $workers = get_targets('workers') + $workers.apply_prep + apply($workers) { + include apt + package { 'openjdk-11-jdk': } + package { 'rsync': } + package { 'vim': } + package { 'less': } + package { 'htop': } + package { 'procps': } + package { 'unzip': } + package { 'sysstat': } + package { 'make': } + package { 'gcc': } + alternatives { java: path => '/usr/lib/jvm/java-11-openjdk-amd64/bin/java' } + } +} diff --git a/benchmark/modules/my/plans/install_agent.pp b/benchmark/modules/my/plans/install_agent.pp new file mode 100644 index 000000000..333c52e45 --- /dev/null +++ b/benchmark/modules/my/plans/install_agent.pp @@ -0,0 +1,3 @@ +plan my::install_agent() { + get_targets("all").apply_prep +} diff --git a/benchmark/modules/my/plans/install_deadsnakes.pp b/benchmark/modules/my/plans/install_deadsnakes.pp new file mode 100644 index 000000000..0349878d2 --- /dev/null +++ b/benchmark/modules/my/plans/install_deadsnakes.pp @@ -0,0 +1,9 @@ +plan my::install_deadsnakes() { + $manager = get_targets('managers') + $manager.apply_prep + apply($manager) { + package { 'lsb-core': } + include apt + apt::ppa { 'ppa:deadsnakes/ppa': } + } +} diff --git a/benchmark/modules/my/plans/localhost_ssh_config.pp b/benchmark/modules/my/plans/localhost_ssh_config.pp new file mode 100644 index 000000000..203233708 --- /dev/null +++ b/benchmark/modules/my/plans/localhost_ssh_config.pp @@ -0,0 +1,37 @@ +plan my::localhost_ssh_config() { + $all = get_targets("all") + $manager = get_target("managers") + 'localhost'.apply_prep + apply('localhost') { + file { "${facts['home']}/.ssh/config": + ensure => present, + } + fm_prepend { "Include flamestream-benchmarks.config": + ensure => present, + data => "Include flamestream-benchmarks.config", + path => "${facts['home']}/.ssh/config", + } + ::ssh::client::config::user { system::env('USER'): + target => "${facts['home']}/.ssh/flamestream-benchmarks.config", + ensure => present, + options => Hash($all.map |Target $target| { + [ + "Host ${target.uri}", if ($target.vars['public_ip']) { + { + 'HostName' => $target.vars['public_ip'], + 'User' => 'ubuntu', + 'StrictHostKeyChecking' => no, + } + } else { + { + 'HostName' => $target.vars['private_ip'], + 'User' => 'ubuntu', + 'ProxyJump' => $manager.uri, + 'StrictHostKeyChecking' => no, + } + }, + ] + }), + } + } +} diff --git a/benchmark/modules/my/plans/localhost_ssh_known_hosts.pp b/benchmark/modules/my/plans/localhost_ssh_known_hosts.pp new file mode 100644 index 000000000..be25863bb --- /dev/null +++ b/benchmark/modules/my/plans/localhost_ssh_known_hosts.pp @@ -0,0 +1,16 @@ +plan my::localhost_ssh_known_hosts() { + $all = get_targets("all") + $all.apply_prep + 'localhost'.apply_prep + apply('localhost') { + $all.each |Target $target| { + $pub = $target.facts['/etc/ssh/ssh_host_ecdsa_key_pub'].split(' ') + sshkey { $target.uri: + type => $pub[0], + key => $pub[1], + host_aliases => if $target.vars['public_ip'] { $target.vars['public_ip'] } else { $target.vars['private_ip'] }, + target => "${facts['home']}/.ssh/known_hosts", + } + } + } +} diff --git a/benchmark/modules/my/plans/ssh.pp b/benchmark/modules/my/plans/ssh.pp new file mode 100644 index 000000000..a4c23cc38 --- /dev/null +++ b/benchmark/modules/my/plans/ssh.pp @@ -0,0 +1,24 @@ +plan my::ssh() { + $all = get_targets('all') + apply($all) { + exec { "/usr/bin/test -e /home/ubuntu/.ssh/id_rsa.pub || ssh-keygen -N '' -f /home/ubuntu/.ssh/id_rsa": } + } + $all.apply_prep + apply($all) { + $all.each |Target $target| { + $client_public_key = split($target.facts['ssh_id_rsa_pub'], ' ') + ssh_authorized_key { $client_public_key[2]: + user => 'ubuntu', + type => $client_public_key[0], + key => $client_public_key[1], + } + $server_public_key = $target.facts['/etc/ssh/ssh_host_ecdsa_key_pub'].split(' ') + sshkey { $target.name: + type => $server_public_key[0], + key => $server_public_key[1], + host_aliases => $target.vars['private_ip'], + target => "${facts['home']}/.ssh/known_hosts", + } + } + } +} diff --git a/benchmark/remote.tf b/benchmark/remote.tf new file mode 120000 index 000000000..456e1faee --- /dev/null +++ b/benchmark/remote.tf @@ -0,0 +1 @@ +yandex/remote.tf \ No newline at end of file diff --git a/benchmark/remote_vms.py b/benchmark/remote_vms.py new file mode 100755 index 000000000..9a6d21b9c --- /dev/null +++ b/benchmark/remote_vms.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +import json +import sys +import subprocess +from tempfile import NamedTemporaryFile +import yaml + +if len(sys.argv) != 2: + raise Exception("exactly 1 argument is expected") + +terraform_run = subprocess.run(["terraform", "output", "--json"], capture_output=True) +terraform_run.check_returncode() +terraform_output = json.loads(terraform_run.stdout) +manager_public_ip = terraform_output['manager_public_ip']['value'].strip() +manager_private_ip = terraform_output['manager_private_ip']['value'].strip() +worker_private_ips = [ip.strip() for ip in terraform_output['worker_private_ips']['value']] + +with NamedTemporaryFile() as inventory_temporary_file: + with open(inventory_temporary_file.name, 'w') as inventory: + inventory.write(yaml.dump(dict(groups=[ + dict(name='managers', targets=[dict( + uri='flamestream-benchmarks-manager', + name='manager', + vars=dict(public_ip=manager_public_ip, private_ip=manager_private_ip), + )]), + dict( + name='workers', + targets=[ + dict( + uri=f"flamestream-benchmarks-worker-{index}", + name=f"worker-{index}", + vars=dict(private_ip=private_ip), + ) + for index, private_ip in enumerate(worker_private_ips) + ], + # config=dict(transport='ssh', ssh={'run-as-command': [ + # 'sudo', + # f'http_proxy=http://{manager_private_ip}:3128', + # f'https_proxy=http://{manager_private_ip}:3128', + # "-nkSEu", + # ]}), + ), + ]))) + + + def bolt_plan_run(*args): + subprocess.run(( + "bolt", "plan", "run", f'--inventoryfile={inventory_temporary_file.name}', "--concurrency=10", *args + )).check_returncode() + + + if sys.argv[1] == "apply": + bolt_plan_run("my::localhost_ssh_config") + bolt_plan_run("my::install_agent", "--run-as=root") + bolt_plan_run("my::localhost_ssh_known_hosts") + # bolt_plan_run("my::http_proxy", "--run-as=root") + bolt_plan_run("my::hosts", "--run-as=root") + bolt_plan_run("my::install_deadsnakes", "--run-as=root") + bolt_plan_run("my::install", "--run-as=root") + bolt_plan_run("my::ssh") + bolt_plan_run("my::bench") + elif sys.argv[1] == "destroy": + bolt_plan_run("my::destroy") + else: + raise Exception("unknown argument") diff --git a/benchmark/terraform.example.tfvars b/benchmark/terraform.example.tfvars new file mode 100644 index 000000000..f5e1a2695 --- /dev/null +++ b/benchmark/terraform.example.tfvars @@ -0,0 +1,7 @@ +access_key = "" +secret_key = "" +google_project = "" +yandex_cloud_id = "" +yandex_folder_id = "" +yandex_token = "" +subnet_id = "" diff --git a/benchmark/yandex/remote.tf b/benchmark/yandex/remote.tf new file mode 100644 index 000000000..e4ff6dedf --- /dev/null +++ b/benchmark/yandex/remote.tf @@ -0,0 +1,78 @@ +variable "yandex_cloud_id" {} +variable "yandex_folder_id" {} +variable "yandex_token" {} +variable "subnet_id" {} + +provider "yandex" { + cloud_id = var.yandex_cloud_id + folder_id = var.yandex_folder_id + token = var.yandex_token + zone = "ru-central1-a" +} + +resource "yandex_compute_instance" "manager" { + name = "flamestream-manager" + boot_disk { + initialize_params { + name = "flamestream-manager" + image_id = "fd8vmcue7aajpmeo39kk" + size = 15 + type = "network-ssd" + } + } + + resources { + cores = 2 + memory = 8 + gpus = 0 + } + metadata = { + ssh-keys = "ubuntu:${file("~/.ssh/id_rsa.pub")}" + } + network_interface { + subnet_id = var.subnet_id + nat = true + } + scheduling_policy { + preemptible = true + } +} + +resource "yandex_compute_instance" "worker" { + count = 5 + name = "flamestream-worker-${count.index}" + boot_disk { + initialize_params { + name = "flamestream-worker-${count.index}" + image_id = "fd8vmcue7aajpmeo39kk" + size = 15 + type = "network-ssd" + } + } + + resources { + cores = 2 + memory = 8 + gpus = 0 + } + metadata = { + ssh-keys = "ubuntu:${file("~/.ssh/id_rsa.pub")}" + } + network_interface { + subnet_id = var.subnet_id + } + scheduling_policy { + preemptible = true + } + allow_stopping_for_update = true +} + +output "manager_public_ip" { + value = yandex_compute_instance.manager.network_interface[0].nat_ip_address +} +output "manager_private_ip" { + value = yandex_compute_instance.manager.network_interface[0].ip_address +} +output "worker_private_ips" { + value = yandex_compute_instance.worker.*.network_interface.0.ip_address +} diff --git a/core/src/main/java/com/spbsu/flamestream/core/Batch.java b/core/src/main/java/com/spbsu/flamestream/core/Batch.java index 669719444..f0e036a21 100644 --- a/core/src/main/java/com/spbsu/flamestream/core/Batch.java +++ b/core/src/main/java/com/spbsu/flamestream/core/Batch.java @@ -2,6 +2,9 @@ import com.spbsu.flamestream.core.data.meta.GlobalTime; +import java.time.Instant; +import java.util.Collections; +import java.util.Map; import java.util.stream.Stream; public interface Batch { @@ -9,6 +12,8 @@ public interface Batch { Stream payload(); + Map lastGlobalTimeProcessedAt(); + enum Default implements Batch { EMPTY; @@ -21,5 +26,10 @@ public GlobalTime time() { public Stream payload() { return Stream.empty(); } + + @Override + public Map lastGlobalTimeProcessedAt() { + return Collections.emptyMap(); + } } } diff --git a/core/src/main/java/com/spbsu/flamestream/core/Graph.java b/core/src/main/java/com/spbsu/flamestream/core/Graph.java index 872408d1e..4b1f606d8 100644 --- a/core/src/main/java/com/spbsu/flamestream/core/Graph.java +++ b/core/src/main/java/com/spbsu/flamestream/core/Graph.java @@ -34,6 +34,8 @@ public interface Graph { Stream adjacent(Vertex vertex); + Stream inverseAdjacent(Vertex vertex); + default TrackingComponent sinkTrackingComponent() { return trackingComponent(sink()); } @@ -102,7 +104,7 @@ public Graph build(Source source, Sink sink) { v.forEach(isolated -> components.add(Collections.singleton(isolated))); - return new MyGraph(adjLists, source, sink, components, vertexTrackingComponentMap, init); + return new MyGraph(adjLists, invertedAdjLists, source, sink, components, vertexTrackingComponentMap, init); } private static class MyGraph implements Graph { @@ -111,12 +113,14 @@ private static class MyGraph implements Graph { private final Sink sink; private final Map> adjLists; + private final Map> invAdjLists; private final Set> components; private final Map vertexTrackingComponent; private final SerializableConsumer init; MyGraph( Multimap adjLists, + Multimap invAdjLists, Source source, Sink sink, Set> components, @@ -124,6 +128,13 @@ private static class MyGraph implements Graph { SerializableConsumer init ) { this.allVertices = new ArrayList<>(adjLists.keySet()); + this.invAdjLists = invAdjLists.entries().stream().collect(Collectors.groupingBy( + Map.Entry::getKey, + Collectors.mapping( + Map.Entry::getValue, + Collectors.toCollection(LinkedHashSet::new) + ) + )); this.components = components; this.vertexTrackingComponent = vertexTrackingComponent; this.init = init; @@ -151,6 +162,11 @@ public Stream adjacent(Vertex vertex) { return adjLists.getOrDefault(vertex, Collections.emptyList()).stream(); } + @Override + public Stream inverseAdjacent(Vertex vertex) { + return invAdjLists.getOrDefault(vertex, Collections.emptyList()).stream(); + } + @Override public TrackingComponent trackingComponent(Vertex vertex) { return vertexTrackingComponent.get(vertex); diff --git a/core/src/main/java/com/spbsu/flamestream/core/data/invalidation/ArrayInvalidatingBucket.java b/core/src/main/java/com/spbsu/flamestream/core/data/invalidation/ArrayInvalidatingBucket.java index 6d0d9e788..a51c97640 100644 --- a/core/src/main/java/com/spbsu/flamestream/core/data/invalidation/ArrayInvalidatingBucket.java +++ b/core/src/main/java/com/spbsu/flamestream/core/data/invalidation/ArrayInvalidatingBucket.java @@ -35,7 +35,7 @@ private ArrayInvalidatingBucket( ) { this.innerList = innerList; this.customOrder = customOrder; - this.order = SerializableComparator.comparing((DataItem dataItem) -> dataItem.meta().globalTime()) + this.order = SerializableComparator.comparingLong((DataItem dataItem) -> dataItem.meta().globalTime().time()) .thenComparing(customOrder) .thenComparing(DataItem::meta); this.dataItemIndexedParents = dataItemIndexedParents; @@ -93,12 +93,12 @@ public boolean isEmpty() { @Override public int insertionPosition(DataItem dataItem) { - return lowerBound(listDataItem -> order.compare(listDataItem, dataItem)); + return lowerBound(listDataItem -> order.compare(dataItem, listDataItem)); } @Override public int lowerBound(GlobalTime globalTime) { - return lowerBound(dataItem -> dataItem.meta().globalTime().compareTo(globalTime)); + return lowerBound(dataItem -> globalTime.compareTo(dataItem.meta().globalTime())); } @Override @@ -124,7 +124,7 @@ private int lowerBound(ToIntFunction comparator) { int right = innerList.size(); while (left != right) { final int middle = left + (right - left) / 2; - if (comparator.applyAsInt(innerList.get(middle)) >= 0) { + if (comparator.applyAsInt(innerList.get(middle)) <= 0) { right = middle; } else { left = middle + 1; diff --git a/core/src/main/java/com/spbsu/flamestream/core/graph/Grouping.java b/core/src/main/java/com/spbsu/flamestream/core/graph/Grouping.java index a6262212c..8f68f71c5 100644 --- a/core/src/main/java/com/spbsu/flamestream/core/graph/Grouping.java +++ b/core/src/main/java/com/spbsu/flamestream/core/graph/Grouping.java @@ -20,6 +20,7 @@ public class Grouping extends HashingVertexStub { private final Class clazz; private final boolean undoPartialWindows; private final SerializableComparator order; + public final boolean timed; public static class Builder { private final HashFunction hash; @@ -28,6 +29,7 @@ public static class Builder { private final Class clazz; private boolean undoPartialWindows = false; private SerializableComparator order = (dataItem1, dataItem2) -> 0; + private boolean timed; public Builder(HashFunction hash, Equalz equalz, int window, Class clazz) { this.window = window; @@ -45,6 +47,11 @@ public Builder order(SerializableComparator order) { this.order = order; return this; } + + public Builder timed(boolean timed) { + this.timed = timed; + return this; + } } public Grouping(HashFunction hash, Equalz equalz, int window, Class clazz) { @@ -54,6 +61,7 @@ public Grouping(HashFunction hash, Equalz equalz, int window, Class clazz) { this.clazz = clazz; undoPartialWindows = false; this.order = (dataItem1, dataItem2) -> 0; + timed = false; } public Grouping(Builder builder) { @@ -63,6 +71,7 @@ public Grouping(Builder builder) { this.clazz = builder.clazz; this.undoPartialWindows = builder.undoPartialWindows; this.order = builder.order; + timed = builder.timed; } public HashFunction hash() { diff --git a/core/src/main/java/com/spbsu/flamestream/core/graph/HashGroup.java b/core/src/main/java/com/spbsu/flamestream/core/graph/HashGroup.java index 17f11585c..3f828a271 100644 --- a/core/src/main/java/com/spbsu/flamestream/core/graph/HashGroup.java +++ b/core/src/main/java/com/spbsu/flamestream/core/graph/HashGroup.java @@ -34,6 +34,15 @@ public boolean covers(int hash) { return false; } + public boolean isEmpty() { + for (final HashUnit unit : units) { + if (!unit.isEmpty()) { + return false; + } + } + return true; + } + @Override public boolean equals(Object o) { if (this == o) { diff --git a/core/src/main/java/com/spbsu/flamestream/core/graph/SerializableComparator.java b/core/src/main/java/com/spbsu/flamestream/core/graph/SerializableComparator.java index 49d642749..8bf19ca87 100644 --- a/core/src/main/java/com/spbsu/flamestream/core/graph/SerializableComparator.java +++ b/core/src/main/java/com/spbsu/flamestream/core/graph/SerializableComparator.java @@ -12,6 +12,13 @@ static > SerializableComparator comparing( return (c1, c2) -> keyExtractor.apply(c1).compareTo(keyExtractor.apply(c2)); } + static SerializableComparator comparingLong( + SerializableToLongFunction keyExtractor + ) { + Objects.requireNonNull(keyExtractor); + return (c1, c2) -> Long.compare(keyExtractor.applyAsLong(c1), keyExtractor.applyAsLong(c2)); + } + default SerializableComparator thenComparing(SerializableComparator other) { Objects.requireNonNull(other); return (c1, c2) -> { diff --git a/examples/nexmark b/examples/nexmark new file mode 160000 index 000000000..4d952a87a --- /dev/null +++ b/examples/nexmark @@ -0,0 +1 @@ +Subproject commit 4d952a87a8beb45b9057ad45519117a80a389aed diff --git a/examples/pom.xml b/examples/pom.xml index a728f9f34..1540186db 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -76,6 +76,45 @@ jblas 1.2.4 + + com.github.nexmark + nexmark-flink + 0.1-SNAPSHOT + + + org.apache.flink + flink-table-api-java + 1.11.1 + compile + + + org.apache.flink + flink-table-api-java-bridge_2.11 + 1.11.1 + compile + + + org.apache.flink + flink-core + 1.11.1 + compile + + + com.esotericsoftware.kryo + kryo + + + org.objenesis + objenesis + + + + + org.apache.flink + flink-streaming-java_2.11 + 1.11.1 + compile + diff --git a/examples/src/main/java/com/spbsu/flamestream/example/benchmark/BenchStandComponentFactory.java b/examples/src/main/java/com/spbsu/flamestream/example/benchmark/BenchStandComponentFactory.java index f74a2790a..9af6f0267 100644 --- a/examples/src/main/java/com/spbsu/flamestream/example/benchmark/BenchStandComponentFactory.java +++ b/examples/src/main/java/com/spbsu/flamestream/example/benchmark/BenchStandComponentFactory.java @@ -7,6 +7,8 @@ import com.spbsu.flamestream.core.data.PayloadDataItem; import com.spbsu.flamestream.core.data.meta.EdgeId; import com.spbsu.flamestream.core.data.meta.GlobalTime; +import com.spbsu.flamestream.core.data.meta.Label; +import com.spbsu.flamestream.core.data.meta.Labels; import com.spbsu.flamestream.core.data.meta.Meta; import com.spbsu.flamestream.runtime.FlameRuntime; import com.spbsu.flamestream.runtime.LocalClusterRuntime; @@ -29,8 +31,6 @@ import java.io.Closeable; import java.io.IOException; import java.io.Writer; -import java.net.InetSocketAddress; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -117,6 +117,9 @@ public Server consumer(Consumer consumer, int port, Class... classesT server.getKryo().register(clazz); } server.getKryo().register(PayloadDataItem.class); + server.getKryo().register(Label.class); + server.getKryo().register(Label[].class); + server.getKryo().register(Labels.class); server.getKryo().register(Meta.class); server.getKryo().register(Meta.ChildIds.class); server.getKryo().register(GlobalTime.class); diff --git a/examples/src/main/java/com/spbsu/flamestream/example/labels/Materializer.java b/examples/src/main/java/com/spbsu/flamestream/example/labels/Materializer.java index 7417334af..93aecfe9b 100644 --- a/examples/src/main/java/com/spbsu/flamestream/example/labels/Materializer.java +++ b/examples/src/main/java/com/spbsu/flamestream/example/labels/Materializer.java @@ -265,7 +265,7 @@ > Graph.Vertex processGrouping(Operator.Groupin grouping.window, typeClass ).order(SerializableComparator.comparing(dataItem -> keyed.order.apply(dataItem.payload(typeClass)))) - .undoPartialWindows(grouping.undoPartialWindows) + .undoPartialWindows(grouping.undoPartialWindows).timed(keyed.timed) ); cachedOperatorVertex.put(grouping, vertex); vertexTrackingComponent.put(vertex, operatorTrackingComponent.get(grouping)); diff --git a/examples/src/main/java/com/spbsu/flamestream/example/labels/Operator.java b/examples/src/main/java/com/spbsu/flamestream/example/labels/Operator.java index a35b9c56e..a04e59f15 100644 --- a/examples/src/main/java/com/spbsu/flamestream/example/labels/Operator.java +++ b/examples/src/main/java/com/spbsu/flamestream/example/labels/Operator.java @@ -4,6 +4,7 @@ import com.spbsu.flamestream.core.graph.SerializableFunction; import com.spbsu.flamestream.core.graph.SerializablePredicate; import com.spbsu.flamestream.core.graph.SerializableToIntFunction; +import org.jetbrains.annotations.Nullable; import scala.Tuple2; import scala.util.Either; @@ -14,6 +15,7 @@ import java.util.Objects; import java.util.Set; import java.util.function.Function; +import java.util.function.ToLongFunction; import java.util.stream.Stream; public abstract class Operator { @@ -79,6 +81,7 @@ public class KeyedBuilder> { private Set> keyLabels = Collections.emptySet(); private Hashing hash = Objects::hashCode; private boolean orderedByProcessingTime = false; + private boolean timed = false; private KeyedBuilder(SerializableFunction key, SerializableFunction order) { this.keyFunction = key; @@ -104,6 +107,11 @@ public KeyedBuilder keyLabels(Set> keyLabels) { return this; } + public KeyedBuilder timed(boolean timed) { + this.timed = timed; + return this; + } + public Keyed build() { return new Keyed<>(this); } @@ -162,6 +170,7 @@ public static final class Keyed> { public final Key> key; public final Hashing hash; public final boolean orderedByProcessingTime; + public final boolean timed; public Keyed(Operator.KeyedBuilder builder) { source = builder.operator(); @@ -179,6 +188,7 @@ public SerializableFunction function() { }; hash = builder.hash; orderedByProcessingTime = builder.orderedByProcessingTime; + timed = builder.timed; } public Operator statefulMap( diff --git a/examples/src/main/java/com/spbsu/flamestream/example/nexmark/BenchStand.java b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/BenchStand.java new file mode 100644 index 000000000..19e061113 --- /dev/null +++ b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/BenchStand.java @@ -0,0 +1,162 @@ +package com.spbsu.flamestream.example.nexmark; + +import com.github.nexmark.flink.NexmarkConfiguration; +import com.spbsu.flamestream.core.DataItem; +import com.spbsu.flamestream.core.data.meta.EdgeId; +import com.spbsu.flamestream.example.benchmark.BenchStandComponentFactory; +import com.spbsu.flamestream.example.benchmark.FlameGraphDeployer; +import com.spbsu.flamestream.example.benchmark.GraphDeployer; +import com.spbsu.flamestream.example.labels.Materializer; +import com.spbsu.flamestream.runtime.config.SystemConfig; +import com.spbsu.flamestream.runtime.edge.socket.SocketRearType; +import com.spbsu.flamestream.runtime.utils.AwaitCountConsumer; +import com.spbsu.flamestream.runtime.utils.tracing.Tracing; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.file.Files; +import java.nio.file.Paths; +import java.text.NumberFormat; +import java.time.Instant; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public class BenchStand { + + private final String workerIdPrefix; + + public static void main(String[] args) throws Exception { + final Config benchConfig; + final Config deployerConfig; + if (args.length == 2) { + benchConfig = ConfigFactory.parseReader(Files.newBufferedReader(Paths.get(args[0]))).getConfig("benchmark"); + deployerConfig = ConfigFactory.parseReader(Files.newBufferedReader(Paths.get(args[1]))).getConfig("deployer"); + } else { + benchConfig = ConfigFactory.load("bench.conf").getConfig("benchmark"); + deployerConfig = ConfigFactory.load("deployer.conf").getConfig("deployer"); + } + final BenchStandComponentFactory benchStandComponentFactory = new BenchStandComponentFactory(); + final var benchStand = new BenchStand(benchConfig, 5); + final var baseTime = System.currentTimeMillis(); + try ( + GraphDeployer graphDeployer = new FlameGraphDeployer( + benchStandComponentFactory.runtime( + deployerConfig, + new SystemConfig.Builder().workersResourcesDistributor( + new SystemConfig.WorkersResourcesDistributor.Enumerated( + benchStand.workerIdPrefix, + 0 + ) + ).build() + ), + Materializer.materialize(Query8.create(benchStand.nexmarkConfiguration.windowSizeSec)), + new GeneratorFrontType( + benchStand.nexmarkConfiguration, + IntStream.range(0, benchStand.parallelism).boxed().collect(Collectors.toMap( + integer -> benchStand.workerIdPrefix + (integer + 1), + Function.identity() + )), + baseTime, + benchStand.maxEvents + ), + new TimingsRearType<>( + new SocketRearType( + benchStand.benchHost, + benchStand.rearPort, + TimingsRearType.Timings.class + ), + 10, + Instant.ofEpochMilli(baseTime) + ) + ) + ) { + benchStand.run(graphDeployer, benchConfig.getString("worker-id")); + } + System.exit(0); + } + + private static final Logger LOG = LoggerFactory.getLogger(BenchStand.class); + + private final double sleepBetweenDocs; + private final String wikiDumpPath; + private final String validatorClass; + public final String benchHost; + private final String inputHost; + public final int frontPort; + public final int rearPort; + private final int parallelism; + public final long maxEvents = 1000000; + public final NexmarkConfiguration nexmarkConfiguration = new NexmarkConfiguration(); + + { + nexmarkConfiguration.bidProportion = 1; + nexmarkConfiguration.personProportion = 1; + nexmarkConfiguration.auctionProportion = 2; + } + + public BenchStand(Config benchConfig, int parallelism) { + sleepBetweenDocs = benchConfig.getDouble("sleep-between-docs-ms"); + wikiDumpPath = benchConfig.getString("wiki-dump-path"); + validatorClass = benchConfig.getString("validator"); + benchHost = benchConfig.getString("bench-host"); + inputHost = benchConfig.getString("input-host"); + frontPort = benchConfig.getInt("bench-source-port"); + rearPort = benchConfig.getInt("bench-sink-port"); + this.parallelism = parallelism; + nexmarkConfiguration.nextEventRate /= parallelism; + nexmarkConfiguration.firstEventRate /= parallelism; + workerIdPrefix = benchConfig.getString("worker-id-prefix"); + } + + private class Window { + int left = parallelism; + long processed = Long.MIN_VALUE; + long notified = Long.MIN_VALUE; + + synchronized void handle(TimingsRearType.Timings timings) { + processed = Long.max(processed, timings.processed); + notified = Long.max(notified, timings.notified); + if (--left == 0) { + System.out.println(format(timings.window) + " " + format(processed) + " " + format(notified)); + } + } + + private String format(long number) { + return NumberFormat.getNumberInstance(Locale.US).format(number); + } + } + + public void run(GraphDeployer graphDeployer, String inputHostId) throws Exception { + final BenchStandComponentFactory benchStandComponentFactory = new BenchStandComponentFactory(); + final var expectedSize = + parallelism * (int) (10 + (maxEvents - 1) / nexmarkConfiguration.firstEventRate / parallelism / nexmarkConfiguration.windowSizeSec + 1); + System.out.println(expectedSize); + final AwaitCountConsumer awaitConsumer = new AwaitCountConsumer( + expectedSize + ); + final Map windows = new ConcurrentHashMap<>(); + try ( + AutoCloseable ignored1 = benchStandComponentFactory.consumer( + x -> { + if (x instanceof DataItem) { + final var timings = ((DataItem) x).payload(TimingsRearType.Timings.class); + awaitConsumer.accept(timings); + windows.computeIfAbsent(timings.window, __ -> new Window()).handle(timings); + } + }, + rearPort, TimingsRearType.Timings.class + )::stop + ) { + graphDeployer.deploy(); + awaitConsumer.await(5, TimeUnit.MINUTES); + Tracing.TRACING.flush(Paths.get("/tmp/trace.csv")); + } + } +} diff --git a/examples/src/main/java/com/spbsu/flamestream/example/nexmark/GeneratorFrontType.java b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/GeneratorFrontType.java new file mode 100644 index 000000000..52924f319 --- /dev/null +++ b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/GeneratorFrontType.java @@ -0,0 +1,167 @@ +package com.spbsu.flamestream.example.nexmark; + +import com.github.nexmark.flink.NexmarkConfiguration; +import com.github.nexmark.flink.generator.GeneratorConfig; +import com.github.nexmark.flink.generator.NexmarkGenerator; +import com.spbsu.flamestream.core.data.PayloadDataItem; +import com.spbsu.flamestream.core.data.meta.GlobalTime; +import com.spbsu.flamestream.core.data.meta.Meta; +import com.spbsu.flamestream.runtime.FlameRuntime; +import com.spbsu.flamestream.runtime.edge.EdgeContext; +import com.spbsu.flamestream.runtime.master.acker.api.Heartbeat; +import com.spbsu.flamestream.runtime.master.acker.api.registry.UnregisterFront; +import org.apache.commons.lang3.SerializationUtils; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.Executors; +import java.util.function.Consumer; + +public class GeneratorFrontType implements FlameRuntime.FrontType { + private final NexmarkConfiguration nexmarkConfiguration; + private final Map nodePartition; + private final long baseTime; + private final long maxEvents; + + public GeneratorFrontType( + NexmarkConfiguration nexmarkConfiguration, + Map nodePartition, + long baseTime, + long maxEvents + ) { + this.nexmarkConfiguration = nexmarkConfiguration; + this.nodePartition = nodePartition; + this.baseTime = baseTime; + this.maxEvents = maxEvents; + } + + public class Instance implements FlameRuntime.FrontInstance { + @Override + public Class clazz() { + return Front.class; + } + + @Override + public Object[] params() { + return new Object[]{GeneratorFrontType.this}; + } + } + + public static class Front implements com.spbsu.flamestream.runtime.edge.Front { + private final EdgeContext edgeContext; + @org.jetbrains.annotations.NotNull + private final GeneratorFrontType type; + private final Integer partition; + + public Front(EdgeContext edgeContext, GeneratorFrontType type) { + this.edgeContext = edgeContext; + this.type = type; + partition = type.nodePartition.get(edgeContext.edgeId().nodeId()); + } + + @Override + public void onStart(Consumer consumer, GlobalTime from) { + if (partition == null) { + consumer.accept(new UnregisterFront(edgeContext.edgeId())); + return; + } + final var executor = + Executors.newSingleThreadExecutor(runnable -> new Thread(runnable, edgeContext.edgeId().toString())); + executor.submit(new Callable() { + private int childId; + private Meta basicMeta; + + @Override + public Object call() throws Exception { + try { + final var warmUpRateReduction = 10; + generate( + generate( + 1, + type.baseTime, + type.nexmarkConfiguration.windowSizeSec * type.nexmarkConfiguration.firstEventRate * type.nodePartition.size(), + slower(type.nexmarkConfiguration, warmUpRateReduction) + ), + type.baseTime + type.nexmarkConfiguration.windowSizeSec * warmUpRateReduction * 1000, + type.maxEvents, + type.nexmarkConfiguration + ); + consumer.accept(new Heartbeat(new GlobalTime(Long.MAX_VALUE, edgeContext.edgeId()))); + return null; + } catch (Throwable throwable) { + throwable.printStackTrace(); + throw throwable; + } finally { + executor.shutdown(); + } + } + + private NexmarkConfiguration slower(NexmarkConfiguration nexmarkConfiguration, int times) { + nexmarkConfiguration = SerializationUtils.clone(nexmarkConfiguration); + nexmarkConfiguration.firstEventRate /= times; + nexmarkConfiguration.nextEventRate /= times; + return nexmarkConfiguration; + } + + private long generate( + long firstEventId, + long baseTime, + long maxEvents, + NexmarkConfiguration nexmarkConfiguration + ) throws InterruptedException { + final var generatorConfig = new GeneratorConfig( + nexmarkConfiguration, + baseTime, + firstEventId, + maxEvents, + 1 + ); + final var generator = new NexmarkGenerator(generatorConfig.split(type.nodePartition.size()).get(partition)); + while (generator.hasNext()) { + final var nextEvent = generator.nextEvent(); + final var event = nextEvent.event; + final var instant = Instant.ofEpochMilli(nextEvent.eventTimestamp); + final var sleep = Duration.between(Instant.now(), instant); + if (!sleep.isNegative()) { + Thread.sleep(sleep.toMillis()); + } + final var time = Query8.tumbleStart(instant, nexmarkConfiguration.windowSizeSec) + - Query8.tumbleStart(Instant.ofEpochMilli(type.baseTime), nexmarkConfiguration.windowSizeSec); + if (basicMeta != null && basicMeta.globalTime().time() > time) { + throw new IllegalArgumentException(); + } + if (basicMeta == null || basicMeta.globalTime().time() < time) { + final var globalTime = new GlobalTime(time, edgeContext.edgeId()); + consumer.accept(new Heartbeat(globalTime)); + basicMeta = new Meta(globalTime); + childId = 0; + } + consumer.accept(new PayloadDataItem(new Meta(basicMeta, 0, childId++), event)); + } + System.out.println("hi"); + return generatorConfig.getStopEventId(); + } + }); + } + + @Override + public void onRequestNext() { + } + + @Override + public void onCheckpoint(GlobalTime to) { + } + } + + @Override + public FlameRuntime.FrontInstance instance() { + return new Instance(); + } + + @Override + public Void handle(EdgeContext context) { + return null; + } +} diff --git a/examples/src/main/java/com/spbsu/flamestream/example/nexmark/Query8.java b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/Query8.java new file mode 100644 index 000000000..ffa52c796 --- /dev/null +++ b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/Query8.java @@ -0,0 +1,200 @@ +package com.spbsu.flamestream.example.nexmark; + +import com.github.nexmark.flink.model.Auction; +import com.github.nexmark.flink.model.Event; +import com.github.nexmark.flink.model.Person; +import com.google.common.hash.Hashing; +import com.spbsu.flamestream.core.graph.SerializableFunction; +import com.spbsu.flamestream.example.labels.Flow; +import com.spbsu.flamestream.example.labels.Operator; +import org.jetbrains.annotations.NotNull; +import scala.Tuple2; +import scala.runtime.AbstractFunction1; +import scala.util.Either; + +import java.time.Instant; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public class Query8 { + public static final class PersonGroupingKey { + public final long id; + public final String name; + public final long startTime; + + private PersonGroupingKey(long id, String name, long startTime) { + this.id = id; + this.name = name; + this.startTime = startTime; + } + + @Override + public int hashCode() { + return Objects.hash(id, name, startTime); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj instanceof PersonGroupingKey) { + final var that = (PersonGroupingKey) obj; + return Objects.equals(id, that.id) && Objects.equals(name, that.name) && startTime == that.startTime; + } + return false; + } + + @Override + public String toString() { + return "(" + id + ", " + name + ", " + startTime + ")"; + } + } + + private static final class AuctionGroupingKey { + public final long seller; + public final long startTime; + + private AuctionGroupingKey(long seller, long startTime) { + this.seller = seller; + this.startTime = startTime; + } + + @Override + public int hashCode() { + return Objects.hash(seller, startTime); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj instanceof AuctionGroupingKey) { + final var that = (AuctionGroupingKey) obj; + return seller == that.seller && startTime == that.startTime; + } + return false; + } + + @Override + public String toString() { + return "(" + seller + ", " + startTime + ")"; + } + } + + private static final class HashedId { + final long id; + + private HashedId(long id) {this.id = id;} + + @Override + public int hashCode() { + return Hashing.murmur3_32().hashLong(id).asInt(); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj instanceof HashedId) { + return id == ((HashedId) obj).id; + } + return false; + } + + @Override + public String toString() { + return "" + id; + } + } + + public static Flow create(long interval) { + final var inputs = new Operator.Input<>(Event.class); + return new Flow<>(inputs, selectJoinOn( + selectUniqueInWindow( + PersonGroupingKey.class, + person -> new PersonGroupingKey(person.id, person.name, tumbleStart(person.dateTime, interval)), + inputs.flatMap(Person.class, event -> Stream.ofNullable(event.newPerson)) + ), + selectUniqueInWindow( + AuctionGroupingKey.class, + auction -> new AuctionGroupingKey(auction.seller, tumbleStart(auction.dateTime, interval)), + inputs.flatMap(Auction.class, event -> Stream.ofNullable(event.newAuction)) + ), + p -> new HashedId(p.id), + a -> new HashedId(a.seller) + )); + } + + private static final class ProcessingOrder implements Comparable { + public static final ProcessingOrder INSTANCE = new ProcessingOrder(); + + private ProcessingOrder() { + } + + @Override + public int compareTo(@NotNull ProcessingOrder o) { + return 1; + } + } + + private static Operator selectUniqueInWindow( + Class outputClass, + SerializableFunction select, + Operator from + ) { + return new Operator.Grouping<>( + from.map(outputClass, select) + .newKeyedBuilder(item -> item, __ -> ProcessingOrder.INSTANCE).timed(true).build(), + 2, + false + ).flatMap(outputClass, group -> group.size() == 1 ? Stream.of(group.get(0)) : Stream.empty()); + } + + private static Operator, List>> selectJoinOn( + Operator leftFrom, + Operator rightFrom, + SerializableFunction leftOn, + SerializableFunction rightOn + ) { + final var eitherClass = (Class>) (Class) Either.class; + final var inputLabel = new Operator.Input<>(eitherClass) + .link(leftFrom.map(eitherClass, scala.util.Left::new)) + .link(rightFrom.map(eitherClass, scala.util.Right::new)); + return new Operator.Grouping<>( + inputLabel.newKeyedBuilder( + either -> either.fold(toScala(leftOn), toScala(rightOn)), + __ -> ProcessingOrder.INSTANCE + ).timed(true).build(), + Integer.MAX_VALUE, + true + ).flatMap((Class, List>>) (Class) Tuple2.class, group -> + group.stream().anyMatch(Either::isLeft) && group.stream().anyMatch(Either::isRight) ? + Stream.of(new Tuple2<>( + group.stream().flatMap(either -> + either.fold(toScala(Stream::of), toScala(__ -> Stream.empty())) + ).collect(Collectors.toList()), + group.stream().flatMap(either -> + either.fold(toScala(__ -> Stream.empty()), toScala(Stream::of)) + ).collect(Collectors.toList()) + )) : Stream.empty() + ); + } + + private static AbstractFunction1 toScala(SerializableFunction function) { + return new AbstractFunction1<>() { + @Override + public Out apply(In in) { + return function.apply(in); + } + }; + } + + public static long tumbleStart(Instant dateTime, long interval) { + return (dateTime.getEpochSecond() / interval + 1) * interval; + } +} diff --git a/examples/src/main/java/com/spbsu/flamestream/example/nexmark/TimingsRearType.java b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/TimingsRearType.java new file mode 100644 index 000000000..856388066 --- /dev/null +++ b/examples/src/main/java/com/spbsu/flamestream/example/nexmark/TimingsRearType.java @@ -0,0 +1,138 @@ +package com.spbsu.flamestream.example.nexmark; + +import com.spbsu.flamestream.core.Batch; +import com.spbsu.flamestream.core.data.PayloadDataItem; +import com.spbsu.flamestream.core.data.meta.Meta; +import com.spbsu.flamestream.runtime.FlameRuntime; +import com.spbsu.flamestream.runtime.edge.EdgeContext; +import com.spbsu.flamestream.runtime.graph.SinkJoba; +import org.apache.commons.lang.ArrayUtils; + +import java.lang.reflect.InvocationTargetException; +import java.text.NumberFormat; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.Collections; +import java.util.Locale; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +public class TimingsRearType + implements FlameRuntime.RearType { + public static class Timings { + public final long window; + public final String nodeId; + public final long processed; + public final long notified; + + public Timings(long window, String nodeId, long processed, long notified) { + this.window = window; + this.nodeId = nodeId; + this.processed = processed; + this.notified = notified; + } + + @Override + public String toString() { + return withDigitSeparators(window) + " seconds window timings " + nodeId + + ": processed = " + withDigitSeparators(processed) + + ", notified = " + withDigitSeparators(notified) + ; + } + + private static String withDigitSeparators(long nanos) { + return NumberFormat.getNumberInstance(Locale.US).format(nanos); + } + } + + final FlameRuntime.RearType wrapped; + final long windowSizeSec; + private final Instant baseTime; + + public TimingsRearType(FlameRuntime.RearType wrapped, long windowSizeSec, Instant baseTime) { + this.wrapped = wrapped; + this.windowSizeSec = windowSizeSec; + this.baseTime = baseTime; + } + + public class Instance implements FlameRuntime.RearInstance { + private final FlameRuntime.RearInstance wrapped; + private final long windowSizeSec = TimingsRearType.this.windowSizeSec; + + public Instance(FlameRuntime.RearInstance wrapped) {this.wrapped = wrapped;} + + @Override + public Class clazz() { + return TimingsRearType.Rear.class; + } + + @Override + public Object[] params() { + return new Object[]{this, wrapped.params()}; + } + + public TimingsRearType type() { + return TimingsRearType.this; + } + } + + public static class Rear implements com.spbsu.flamestream.runtime.edge.Rear { + private final EdgeContext edgeContext; + @org.jetbrains.annotations.NotNull + private final TimingsRearType.Instance instance; + private final com.spbsu.flamestream.runtime.edge.Rear wrapped; + + public Rear( + EdgeContext edgeContext, TimingsRearType.Instance instance, Object[] params + ) throws IllegalAccessException, InvocationTargetException, InstantiationException { + this.edgeContext = edgeContext; + this.instance = instance; + wrapped = (com.spbsu.flamestream.runtime.edge.Rear) instance.wrapped.clazz().getDeclaredConstructors()[0] + .newInstance(ArrayUtils.add(params, 0, edgeContext)); + } + + Batch last = Batch.Default.EMPTY; + + @Override + public CompletionStage accept(Batch batch) { + if (batch.time().time() < Long.MAX_VALUE) { + final var window = batch.time().time() - instance.windowSizeSec; + final var processed = batch.lastGlobalTimeProcessedAt().get(window); + if (processed == null) { + throw new IllegalStateException(batch.lastGlobalTimeProcessedAt().toString()); + } + final var now = Instant.now(); + final var windowEnd = Instant.ofEpochSecond( + Query8.tumbleStart(instance.type().baseTime.plus(window, ChronoUnit.SECONDS), instance.windowSizeSec) + ); + return wrapped.accept(new SinkJoba.BatchImpl( + batch.time(), + Collections.singletonList(new PayloadDataItem(new Meta(batch.time()), new Timings( + window, + edgeContext.edgeId().nodeId(), + windowEnd.until(processed, ChronoUnit.NANOS), + windowEnd.until(now, ChronoUnit.NANOS) + ))), + batch.lastGlobalTimeProcessedAt() + )); + } + last = batch; + return CompletableFuture.completedStage(null); + } + + @Override + public Batch last() { + return last; + } + } + + @Override + public Instance instance() { + return new Instance(wrapped.instance()); + } + + @Override + public Handle handle(EdgeContext context) { + return wrapped.handle(context); + } +} diff --git a/examples/src/main/resources/deployer.conf b/examples/src/main/resources/deployer.conf index c60a54ad6..17b3b3157 100644 --- a/examples/src/main/resources/deployer.conf +++ b/examples/src/main/resources/deployer.conf @@ -4,7 +4,7 @@ deployer { // } local-cluster { - parallelism = 4 + parallelism = 5 } //remote { diff --git a/examples/src/test/java/com/spbsu/flamestream/example/nexmark/Query8Test.java b/examples/src/test/java/com/spbsu/flamestream/example/nexmark/Query8Test.java new file mode 100644 index 000000000..10b33b71f --- /dev/null +++ b/examples/src/test/java/com/spbsu/flamestream/example/nexmark/Query8Test.java @@ -0,0 +1,53 @@ +package com.spbsu.flamestream.example.nexmark; + +import com.github.nexmark.flink.NexmarkConfiguration; +import com.spbsu.flamestream.core.data.meta.EdgeId; +import com.spbsu.flamestream.example.labels.Materializer; +import com.spbsu.flamestream.runtime.FlameRuntime; +import com.spbsu.flamestream.runtime.LocalRuntime; +import com.spbsu.flamestream.runtime.acceptance.FlameAkkaSuite; +import com.spbsu.flamestream.runtime.config.SystemConfig; +import org.testng.annotations.Test; + +import java.time.Instant; +import java.util.Map; +import java.util.stream.Collectors; + +public class Query8Test extends FlameAkkaSuite { + @Test + public void test() throws Exception { + //var nexmarkConfiguration = new NexmarkConfiguration(); + //nexmarkConfiguration.bidProportion = 1; + //nexmarkConfiguration.personProportion = 1; + //nexmarkConfiguration.auctionProportion = 2; + //nexmarkConfiguration.firstEventRate /= 1; + //nexmarkConfiguration.nextEventRate /= 1; + //nexmarkConfiguration.windowSizeSec *= 1; + // + //final var graph = Materializer.materialize(Query8.create(nexmarkConfiguration.windowSizeSec)); + // + //final var currentTimeMillis = System.currentTimeMillis(); + //try (final var runtime = new LocalRuntime.Builder() + // .millisBetweenCommits(Integer.MAX_VALUE) + // .defaultMinimalTime(Query8.tumbleStart( + // Instant.ofEpochMilli(currentTimeMillis), + // nexmarkConfiguration.windowSizeSec + // )).acking(SystemConfig.Acking.DISABLED) + // .build()) { + // try (final FlameRuntime.Flame flame = runtime.run(graph)) { + // final var rears = + // flame.attachRear("rear", new TimingsRearType<>(new SimpleRearType(), 10)).collect(Collectors.toList()); + // + // flame.attachFront("front", new GeneratorFrontType(nexmarkConfiguration, Map.ofEntries( + // Map.entry("node-0", 0), + // Map.entry("node-1", 1), + // Map.entry("node-2", 2), + // Map.entry("node-3", 3) + // ), currentTimeMillis, 10000)); + // for (final var rear : rears) { + // rear.await(); + // } + // } + //} + } +} diff --git a/pom.xml b/pom.xml index cdba0e223..6256d3671 100644 --- a/pom.xml +++ b/pom.xml @@ -17,5 +17,6 @@ examples client benchmark/flink-benchmark + examples/nexmark/nexmark-flink \ No newline at end of file diff --git a/run_benchmarks.sh b/run_benchmarks.sh index 78c051a9c..738c6bfda 100755 --- a/run_benchmarks.sh +++ b/run_benchmarks.sh @@ -68,4 +68,4 @@ remote_flink_bench() { && ansible-playbook -v -i "${ANSIBLE_HOME}/aws.yml" "${ANSIBLE_HOME}/flink.yml" } -[[ "$0" == "$BASH_SOURCE" ]] && local_flink_bench +[[ "$0" == "$BASH_SOURCE" ]] && package copy_worker_artifacts diff --git a/runtime/entrypoint.sh b/runtime/entrypoint.sh index 2e9cf8bd9..c698aa415 100644 --- a/runtime/entrypoint.sh +++ b/runtime/entrypoint.sh @@ -6,7 +6,7 @@ usage() { start() { echo "Starting flamestream worker" - local java_ops="-Daeron.term.buffer.length=4194304 -Daeron.mtu.length=16384 -Xms500m -Xmx6000m -Xlog:gc,gc+cpu=info::utc -XX:+HeapDumpOnOutOfMemoryError" + local java_ops="-Daeron.term.buffer.length=134217728 -Daeron.mtu.length=16384 -Xms500m -Xmx6000m -Xlog:gc,gc+cpu=info::utc -XX:+HeapDumpOnOutOfMemoryError" echo "java_ops=$java_ops" local main="com.spbsu.flamestream.runtime.WorkerApplication" nohup bash -c "java $java_ops -cp lib/*:flamestream-runtime-1.0-SNAPSHOT.jar $main &>worker.log" & @@ -39,4 +39,4 @@ main() { esac } -main "$@" \ No newline at end of file +main "$@" diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/WorkerApplication.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/WorkerApplication.java index 6eb6d5a45..20899c976 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/WorkerApplication.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/WorkerApplication.java @@ -61,7 +61,7 @@ public static void main(String... args) { configureFromEnv("BARRIER_DISABLED", Boolean::parseBoolean, systemConfigBuilder::barrierDisabled); configureFromEnv("ACKERS_NUMBER", (int ackersNumber) -> systemConfigBuilder.workersResourcesDistributor(new SystemConfig.WorkersResourcesDistributor.Enumerated( - "flamestream-benchmarks-worker-", + "worker-", ackersNumber )) ); diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/config/SystemConfig.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/config/SystemConfig.java index 005523e9b..eeebe62e0 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/config/SystemConfig.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/config/SystemConfig.java @@ -161,7 +161,7 @@ public Props localAckerProps(List ackers, String nodeId) { public static class Builder { private int maxElementsInGraph = 500; private int millisBetweenCommits = 100; - private int defaultMinimalTime = 0; + private long defaultMinimalTime = 0; private boolean barrierDisabled = false; private LocalAcker.Builder localAckerBuilder = new LocalAcker.Builder(); private int ackerWindow = 1; @@ -178,7 +178,7 @@ public Builder millisBetweenCommits(int millisBetweenCommits) { return this; } - public Builder defaultMinimalTime(int defaultMinimalTime) { + public Builder defaultMinimalTime(long defaultMinimalTime) { this.defaultMinimalTime = defaultMinimalTime; return this; } diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/config/ZookeeperWorkersNode.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/config/ZookeeperWorkersNode.java index f63aa651c..43cc04516 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/config/ZookeeperWorkersNode.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/config/ZookeeperWorkersNode.java @@ -7,6 +7,7 @@ import com.spbsu.flamestream.runtime.serialization.JacksonSerializer; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.cache.PathChildrenCache; +import org.apache.curator.utils.ZKPaths; import org.apache.zookeeper.CreateMode; import java.util.List; @@ -39,8 +40,9 @@ static public class Worker { public ZookeeperWorkersNode(CuratorFramework curator, String path) { this.curator = curator; this.path = path; - this.pathChildrenCache = new PathChildrenCache(curator, path, true); try { + ZKPaths.mkdirs(curator.getZookeeperClient().getZooKeeper(), path); + this.pathChildrenCache = new PathChildrenCache(curator, path, true); pathChildrenCache.start(); } catch (Exception e) { throw new RuntimeException(e); diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/edge/SimpleFront.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/edge/SimpleFront.java new file mode 100644 index 000000000..05ee13e79 --- /dev/null +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/edge/SimpleFront.java @@ -0,0 +1,108 @@ +package com.spbsu.flamestream.runtime.edge; + +import com.spbsu.flamestream.core.data.PayloadDataItem; +import com.spbsu.flamestream.core.data.meta.GlobalTime; +import com.spbsu.flamestream.core.data.meta.Meta; +import com.spbsu.flamestream.runtime.FlameRuntime; +import com.spbsu.flamestream.runtime.master.acker.api.Heartbeat; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutionException; +import java.util.function.Consumer; + +public class SimpleFront implements Front { + private final CompletableFuture> consumers; + + public static class Handle { + private final Consumer consumer; + private final EdgeContext context; + private Meta basicMeta; + private int childId = 0; + + public Handle(ConcurrentMap>> consumer, EdgeContext context) { + try { + this.consumer = consumer.computeIfAbsent(context, __ -> new CompletableFuture<>()).get(); + } catch (InterruptedException | ExecutionException e) { + throw new RuntimeException(e); + } + this.context = context; + } + + public synchronized boolean onDataItem(long time, Object object) { + if (basicMeta != null && basicMeta.globalTime().time() > time) { + throw new IllegalArgumentException(); + } + boolean heartbeat = false; + if (basicMeta == null || basicMeta.globalTime().time() < time) { + final var globalTime = new GlobalTime(time, context.edgeId()); + consumer.accept(new Heartbeat(globalTime)); + basicMeta = new Meta(globalTime); + childId = 0; + heartbeat = true; + } + final var dataItem = new PayloadDataItem(new Meta(basicMeta, 0, childId++), object); + consumer.accept(dataItem); + return heartbeat; + } + + public synchronized void unregister() { + consumer.accept(new Heartbeat(new GlobalTime(Long.MAX_VALUE, context.edgeId()))); + } + } + + public static class Instance implements FlameRuntime.FrontInstance { + private final Map>> consumers; + + public Instance(Map>> consumers) {this.consumers = consumers;} + + @Override + public Class clazz() { + return SimpleFront.class; + } + + @Override + public Object[] params() { + return new Object[]{consumers}; + } + } + + public static class Type implements FlameRuntime.FrontType { + private final ConcurrentMap>> consumers + = new ConcurrentHashMap<>(); + + @Override + public Instance instance() { + return new Instance(consumers); + } + + @Override + public Handle handle(EdgeContext context) { + return new Handle(consumers, context); + } + } + + public SimpleFront( + EdgeContext edgeContext, + ConcurrentMap>> consumers + ) { + this.consumers = consumers.computeIfAbsent(edgeContext, __ -> new CompletableFuture<>()); + } + + @Override + public void onStart(Consumer consumer, GlobalTime from) { + if (!consumers.complete(consumer)) { + throw new IllegalStateException(); + } + } + + @Override + public void onRequestNext() { + } + + @Override + public void onCheckpoint(GlobalTime to) { + } +} diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/edge/akka/AkkaRear.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/edge/akka/AkkaRear.java index 72309fba2..b463f60c1 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/edge/akka/AkkaRear.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/edge/akka/AkkaRear.java @@ -32,6 +32,7 @@ public AkkaRear(EdgeContext edgeContext, ActorRefFactory refFactory, String loca @Override public CompletionStage accept(Batch batch) { + System.out.println(batch); return PatternsCS.ask(innerActor, batch, FlameConfig.config.smallTimeout()); } diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Component.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Component.java index 2af1451dc..ebb9d3413 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Component.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Component.java @@ -7,6 +7,7 @@ import com.spbsu.flamestream.core.Graph; import com.spbsu.flamestream.core.HashFunction; import com.spbsu.flamestream.core.TrackingComponent; +import com.spbsu.flamestream.core.data.meta.EdgeId; import com.spbsu.flamestream.core.data.meta.GlobalTime; import com.spbsu.flamestream.core.data.meta.Meta; import com.spbsu.flamestream.core.graph.FlameMap; @@ -21,7 +22,9 @@ import com.spbsu.flamestream.runtime.graph.api.AddressedItem; import com.spbsu.flamestream.runtime.graph.api.ComponentPrepared; import com.spbsu.flamestream.runtime.graph.api.NewRear; +import com.spbsu.flamestream.runtime.graph.api.Watermark; import com.spbsu.flamestream.runtime.graph.state.GroupGroupingState; +import com.spbsu.flamestream.runtime.master.acker.NodeTimes; import com.spbsu.flamestream.runtime.master.acker.api.Ack; import com.spbsu.flamestream.runtime.master.acker.api.Heartbeat; import com.spbsu.flamestream.runtime.master.acker.api.MinTimeUpdate; @@ -34,30 +37,97 @@ import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import java.time.Instant; import java.util.ArrayList; import java.util.Collection; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.NavigableMap; import java.util.Objects; import java.util.Set; +import java.util.TreeMap; import java.util.function.BiConsumer; import java.util.function.Function; import java.util.stream.Collectors; +import java.util.stream.IntStream; public class Component extends LoggingActor { + private final String nodeId; @NotNull private final Graph graph; + private final HashUnitMap routes; + private final ActorRef localManager; private final ActorRef localAcker; + private final ComputationProps props; + private final TrackingComponent sinkTrackingComponent; private class JobaWrapper { WrappedJoba joba; Joba.Sink downstream; final Graph.Vertex vertex; + final Map incomingWatermarks = new HashMap<>(); + long outgoingWatermark; JobaWrapper(WrappedJoba joba, Joba.Sink downstream, Graph.Vertex vertex) { this.joba = joba; this.downstream = downstream; this.vertex = vertex; + final var hash = vertex instanceof HashingVertexStub ? ((HashingVertexStub) vertex).hash() : null; + if (joba instanceof SourceJoba) { + incomingWatermarks.put(null, props.defaultMinimalTime()); + } else if (hash == null) { + graph.inverseAdjacent(vertex).forEach(incoming -> + incomingWatermarks.put(new Joba.Id(nodeId, incoming.id()), props.defaultMinimalTime()) + ); + } else { + for (final var entry : props.hashGroups().entrySet()) { + if (!entry.getValue().isEmpty()) { + graph.inverseAdjacent(vertex).forEach(incoming -> + incomingWatermarks.put(new Joba.Id(entry.getKey(), incoming.id()), props.defaultMinimalTime()) + ); + } + } + } + } + + void onWatermark(Joba.Id from, long time) { + if (incomingWatermarks.put(from, time) == null) { + throw new IllegalStateException(from.toString()); + } + final var outgoingWatermark = incomingWatermarks.values().stream().mapToLong(l -> l).min().orElse(Long.MAX_VALUE); + final var self = new Joba.Id(nodeId, this.vertex.id()); + if (this.outgoingWatermark < outgoingWatermark) { + graph.adjacent(this.vertex).forEach(to -> { + final var destination = GraphManager.Destination.fromVertexId(to.id()); + final var hashing = to instanceof HashingVertexStub ? ((HashingVertexStub) to).hash() : null; + if (hashing == null) { + wrappedJobas.get(destination).onWatermark(self, outgoingWatermark); + } else { + props.hashGroups().forEach((nodeId, hashGroup) -> { + final var hash = hashGroup.units().stream().flatMapToInt(entry -> + entry.isEmpty() ? IntStream.empty() : IntStream.of(entry.from()) + ).findAny(); + if (hash.isPresent()) { + final var route = routes.get(hash.getAsInt()); + if (route == localManager) { + wrappedJobas.get(destination).onWatermark(self, outgoingWatermark); + } else { + route.tell(new Watermark(self, destination, outgoingWatermark), self()); + } + } + }); + } + }); + if (vertex instanceof Sink) { + onMinTime(new MinTimeUpdate( + sinkTrackingComponent.index, + new GlobalTime(outgoingWatermark, EdgeId.MAX), + new NodeTimes() + )); + } + this.outgoingWatermark = outgoingWatermark; + } } } @@ -74,17 +144,23 @@ private class JobaWrapper { @Nullable private JobaWrapper wrappedSinkJoba; - private Component(String nodeId, - Set componentVertices, - Graph graph, - HashUnitMap routes, - ActorRef localManager, - @Nullable ActorRef localAcker, - ComputationProps props, - Map stateByVertex) { + private Component( + String nodeId, + Set componentVertices, + Graph graph, + HashUnitMap routes, + ActorRef localManager, + @Nullable ActorRef localAcker, + ComputationProps props, + Map stateByVertex + ) { + this.nodeId = nodeId; this.graph = graph; + this.routes = routes; + this.localManager = localManager; this.localAcker = localAcker; - final TrackingComponent sinkTrackingComponent = graph.sinkTrackingComponent(); + this.props = props; + sinkTrackingComponent = graph.sinkTrackingComponent(); this.wrappedJobas = componentVertices.stream().collect(Collectors.toMap( vertex -> GraphManager.Destination.fromVertexId(vertex.id()), vertex -> { @@ -239,14 +315,16 @@ public void accept(DataItem item) { )); } - public static Props props(String nodeId, - Set componentVertices, - Graph graph, - HashUnitMap localManager, - ActorRef routes, - @Nullable ActorRef localAcker, - ComputationProps props, - Map stateByVertex) { + public static Props props( + String nodeId, + Set componentVertices, + Graph graph, + HashUnitMap localManager, + ActorRef routes, + @Nullable ActorRef localAcker, + ComputationProps props, + Map stateByVertex + ) { return Props.create( Component.class, nodeId, @@ -270,11 +348,18 @@ public Receive createReceive() { .match(Heartbeat.class, h -> { if (localAcker != null) { localAcker.forward(h, context()); + } else { + wrappedSourceJoba.onWatermark(null, h.time().time()); } }) + .match(Watermark.class, watermark -> + wrappedJobas.get(watermark.to).onWatermark(watermark.from, watermark.time) + ) .match(UnregisterFront.class, u -> { if (localAcker != null) { localAcker.forward(u, context()); + } else { + wrappedSourceJoba.onWatermark(null, Long.MAX_VALUE); } }) .match(Prepare.class, this::onPrepare) @@ -299,6 +384,9 @@ private void inject(AddressedItem addressedItem) { localCall(item, addressedItem.destination()); ack(item, wrappedJobas.get(addressedItem.destination()).vertex); injectOutTracer.log(item.xor()); + if (wrappedSinkJoba != null) { + wrappedSinkJoba.joba.touch(item.meta().globalTime().time()); + } } private void ack(DataItem dataItem, Graph.Vertex to) { @@ -329,6 +417,9 @@ private void accept(DataItem item) { } else { throw new IllegalStateException("Source doesn't belong to this component"); } + if (wrappedSinkJoba != null) { + wrappedSinkJoba.joba.touch(item.meta().globalTime().time()); + } } private void onNewRear(NewRear attachRear) { diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/GraphManager.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/GraphManager.java index f1d77092d..842a9c77b 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/GraphManager.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/GraphManager.java @@ -14,6 +14,7 @@ import com.spbsu.flamestream.core.graph.HashUnit; import com.spbsu.flamestream.runtime.graph.api.AddressedItem; import com.spbsu.flamestream.runtime.graph.api.NewRear; +import com.spbsu.flamestream.runtime.graph.api.Watermark; import com.spbsu.flamestream.runtime.graph.state.GroupGroupingState; import com.spbsu.flamestream.runtime.graph.state.GroupingState; import com.spbsu.flamestream.runtime.master.acker.api.Heartbeat; @@ -40,6 +41,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; public class GraphManager extends LoggingActor { @@ -194,6 +196,7 @@ private Receive managing() { .match(Commit.class, commit -> sourceComponent.forward(commit, context())) .match(NewRear.class, newRear -> sinkComponent.forward(newRear, context())) .match(Heartbeat.class, gt -> sourceComponent.forward(gt, context())) + .match(Watermark.class, watermark -> verticesComponents.get(watermark.to).forward(watermark, context())) .match(UnregisterFront.class, u -> sourceComponent.forward(u, context())) .build(); } @@ -225,7 +228,7 @@ private void onMinTime(MinTimeUpdate minTimeUpdate) { } public static class Destination { - private static final Map cache = new HashMap<>(); + private static final Map cache = new ConcurrentHashMap<>(); private final String vertexId; @Override diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Joba.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Joba.java index c9dcbbab3..e0b6bcc35 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Joba.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/Joba.java @@ -34,6 +34,11 @@ public boolean equals(Object o) { Objects.equals(vertexId, that.vertexId); } + @Override + public String toString() { + return "Joba.Id(" + nodeId + ", " + vertexId + ")"; + } + @Override public int hashCode() { return Objects.hash(nodeId, vertexId); diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/SinkJoba.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/SinkJoba.java index 47ed08e46..a9d0ba50b 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/SinkJoba.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/SinkJoba.java @@ -15,11 +15,14 @@ import com.spbsu.flamestream.runtime.utils.FlameConfig; import com.spbsu.flamestream.runtime.utils.tracing.Tracing; +import java.time.Instant; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.NavigableMap; +import java.util.TreeMap; import java.util.concurrent.ExecutionException; import java.util.stream.Stream; @@ -27,8 +30,9 @@ public class SinkJoba extends Joba { private final LoggingAdapter log; private final boolean barrierDisabled; private final int sinkTrackingComponent; - private final InvalidatingBucket invalidatingBucket = new ArrayInvalidatingBucket(); + private final TreeMap timeInvalidatingBucket = new TreeMap<>(); private final Map rears = new HashMap<>(); + private NavigableMap lastGlobalTimeProcessedAt = new TreeMap<>(); private final Tracing.Tracer barrierReceiveTracer = Tracing.TRACING.forEvent("barrier-receive"); private final Tracing.Tracer barrierSendTracer = Tracing.TRACING.forEvent("barrier-send"); @@ -48,10 +52,12 @@ public void accept(DataItem item, Sink sink) { if (barrierDisabled) { rears.forEach((rear, lastEmmit) -> emmitRearBatch( rear, - new BatchImpl(item.meta().globalTime(), Collections.singletonList(item)) + new BatchImpl(item.meta().globalTime(), Collections.singletonList(item), Collections.emptyMap()) )); } else { - invalidatingBucket.insert(item); + timeInvalidatingBucket + .computeIfAbsent(item.meta().globalTime().time(), __ -> new ArrayInvalidatingBucket((___, ____) -> 1)) + .insert(item); } } @@ -77,28 +83,43 @@ public void onMinTime(MinTimeUpdate minTime) { } } + public void touch(long time) { + lastGlobalTimeProcessedAt.put(time, Instant.now()); + } + private void tryEmmit(GlobalTime upTo) { - final int pos = invalidatingBucket.lowerBound(upTo); - - rears.forEach((rear, lastEmmit) -> { - final List data = new ArrayList<>(); - invalidatingBucket.forRange(0, pos, item -> { - if (item.meta().globalTime().compareTo(lastEmmit) > 0) { - data.add(item); - } + final var iterator = timeInvalidatingBucket.entrySet().iterator(); + while (iterator.hasNext()) { + final var invalidatingBucketEntry = iterator.next(); + if (upTo.time() <= invalidatingBucketEntry.getKey()) { + return; + } + final var invalidatingBucket = invalidatingBucketEntry.getValue(); + final int pos = invalidatingBucket.lowerBound(upTo); + + final var lastGlobalTimeProcessedAt = this.lastGlobalTimeProcessedAt.headMap(upTo.time()); + rears.forEach((rear, lastEmmit) -> { + final List data = new ArrayList<>(); + invalidatingBucket.forRange(0, pos, item -> { + if (item.meta().globalTime().compareTo(lastEmmit) > 0) { + data.add(item); + } + }); + + emmitRearBatch(rear, new BatchImpl(upTo, data, Map.copyOf(lastGlobalTimeProcessedAt))); }); - if (!data.isEmpty() || barrierDisabled) { - emmitRearBatch(rear, new BatchImpl(upTo, data)); + // Clearing barrier only if elements were emitted somewhere. + // It is temporary fix of the "sending elements to /dev/null" problem + // + // https://github.com/flame-stream/FlameStream/issues/139 + if (!rears.isEmpty()) { + invalidatingBucket.clearRange(0, pos); + lastGlobalTimeProcessedAt.clear(); + } + if (invalidatingBucket.isEmpty()) { + iterator.remove(); } - }); - - // Clearing barrier only if elements were emitted somewhere. - // It is temporary fix of the "sending elements to /dev/null" problem - // - // https://github.com/flame-stream/FlameStream/issues/139 - if (!rears.isEmpty()) { - invalidatingBucket.clearRange(0, pos); } } @@ -114,10 +135,17 @@ private void emmitRearBatch(ActorRef rear, BatchImpl batch) { public static class BatchImpl implements Batch { private final List items; private final GlobalTime time; + private final Map lastGlobalTimeProcessedAt; - private BatchImpl(GlobalTime time, List items) { + public BatchImpl(GlobalTime time, List items, Map lastGlobalTimeProcessedAt) { this.items = items; this.time = time; + this.lastGlobalTimeProcessedAt = lastGlobalTimeProcessedAt; + } + + @Override + public String toString() { + return time + " " + items.size(); } @Override @@ -129,5 +157,10 @@ public GlobalTime time() { public Stream payload() { return items.stream(); } + + @Override + public Map lastGlobalTimeProcessedAt() { + return lastGlobalTimeProcessedAt; + } } } diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/api/Watermark.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/api/Watermark.java new file mode 100644 index 000000000..4aa19771d --- /dev/null +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/api/Watermark.java @@ -0,0 +1,21 @@ +package com.spbsu.flamestream.runtime.graph.api; + +import com.spbsu.flamestream.runtime.graph.GraphManager; +import com.spbsu.flamestream.runtime.graph.Joba; + +public class Watermark { + public final Joba.Id from; + public final GraphManager.Destination to; + public final long time; + + public Watermark(Joba.Id from, GraphManager.Destination to, long time) { + this.from = from; + this.to = to; + this.time = time; + } + + @Override + public String toString() { + return "Watermark(" + from + ", " + to + ", " + time + ")"; + } +} diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/state/GroupingState.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/state/GroupingState.java index 1f5291f83..96e3de9b8 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/state/GroupingState.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/graph/state/GroupingState.java @@ -34,6 +34,9 @@ public boolean equals(Object obj) { } if (obj instanceof Key) { final Key key = (Key) obj; + if (grouping.timed && dataItem.meta().globalTime().time() != ((Key) obj).dataItem.meta().globalTime().time()) { + return false; + } return grouping.equalz().test(dataItem, key.dataItem); } return false; @@ -47,9 +50,10 @@ public GroupingState(Grouping grouping) { this.grouping = grouping; map = new TimedMap<>( () -> new SynchronizedInvalidatingBucket(grouping.order()), - grouping.equalz().labels().isEmpty() ? null : key -> grouping.equalz().labels().stream().mapToLong(label -> - key.dataItem.meta().labels().get(label).globalTime.time() - ).min().getAsLong() + grouping.equalz().labels().isEmpty() && !grouping.timed ? null : key -> + grouping.equalz().labels().stream().mapToLong(label -> + key.dataItem.meta().labels().get(label).globalTime.time() + ).min().orElse(grouping.timed ? key.dataItem.meta().globalTime().time() : Long.MAX_VALUE) ); } diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/master/acker/NodeTimes.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/master/acker/NodeTimes.java index 36b498002..993d6e6cb 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/master/acker/NodeTimes.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/master/acker/NodeTimes.java @@ -9,7 +9,7 @@ public class NodeTimes { private final HashMap all; - NodeTimes() { + public NodeTimes() { this.all = new HashMap<>(); } diff --git a/runtime/src/main/java/com/spbsu/flamestream/runtime/utils/AwaitCountConsumer.java b/runtime/src/main/java/com/spbsu/flamestream/runtime/utils/AwaitCountConsumer.java index c8f482720..d038334c3 100644 --- a/runtime/src/main/java/com/spbsu/flamestream/runtime/utils/AwaitCountConsumer.java +++ b/runtime/src/main/java/com/spbsu/flamestream/runtime/utils/AwaitCountConsumer.java @@ -11,6 +11,7 @@ public class AwaitCountConsumer implements Consumer { private final AtomicInteger counter = new AtomicInteger(); private final int expectedSize; + private long acceptedAt = Long.MIN_VALUE; public AwaitCountConsumer(int expectedSize) { this.expectedSize = expectedSize; @@ -18,6 +19,7 @@ public AwaitCountConsumer(int expectedSize) { @Override public void accept(Object o) { + acceptedAt = System.currentTimeMillis(); if (counter.incrementAndGet() == expectedSize) { synchronized (counter) { counter.notifyAll(); diff --git a/runtime/src/main/resources/local.conf b/runtime/src/main/resources/local.conf index dcf92bbc2..866cc1796 100644 --- a/runtime/src/main/resources/local.conf +++ b/runtime/src/main/resources/local.conf @@ -27,6 +27,7 @@ akka { "com.spbsu.flamestream.runtime.master.acker.RegistryHolder$NewFrontRegisterer$Registered" = kryo "com.spbsu.flamestream.runtime.master.acker.RegistryHolder$AlreadyRegisteredFrontRegisterer$Registered" = kryo "com.spbsu.flamestream.runtime.graph.api.AddressedItem" = kryo + "com.spbsu.flamestream.runtime.graph.api.Watermark" = kryo "com.spbsu.flamestream.runtime.master.acker.api.registry.UnregisterFront" = kryo "com.spbsu.flamestream.runtime.master.acker.api.commit.GimmeLastCommit" = kryo "com.spbsu.flamestream.runtime.master.acker.api.commit.LastCommit" = kryo diff --git a/runtime/src/main/resources/remote.conf b/runtime/src/main/resources/remote.conf index cbaa6fcdc..fa12b96f9 100644 --- a/runtime/src/main/resources/remote.conf +++ b/runtime/src/main/resources/remote.conf @@ -4,11 +4,16 @@ akka { remote { artery { enabled = on + transport = tcp advanced { maximum-frame-size = 1024 KiB idle-cpu-level = 1 inbound-lanes = 1 outbound-lanes = 1 + give-up-message-after = 3600 seconds + aeron { + give-up-message-after = 3600 seconds + } } }