Task- und Datenparallelität mit Rust

fn main() {
    // Die Statements werden ausgeführt sobald
    // das compilierte Binary gestartet wird.

    // Ausgabe auf stdout
    println!("Hello para//el 2019!");
}

// A type is `Send` if it is safe to send it to another thread.
// A type is `Sync` if it is safe to share between threads.

fn foo<T:Send+Sync>(value: T) {
	...
}

#[derive(Copy, Clone, Debug)]
pub struct Vector<T> {
	pub x: T,
	pub y: T,
	pub z: T
}

/// Definition des Operators +=
impl<T: Add<Output=T> + AddAssign> AddAssign for Vector<T> {
	fn add_assign(&mut self, other: Vector<T>) {
		self.x += other.x;
		self.y += other.y;
		self.z += other.z;
	}
}

#[derive(Clone, Debug, PartialEq)]
pub struct NBody {
	pub position: Vec<Vector<Precision>>,
	pub velocity: Vec<Vector<Precision>>
}

position.iter().zip(velocity.iter_mut())
  .for_each(|(item_pi, item_vi)| {
	let mut f: Vector<Precision> = Vector::new(0.0, 0.0, 0.0);

	position.iter().for_each(|item_pj| {
		// Newton’s law of universal gravity calculation.
		let diff = *item_pj - *item_pi;
		let n2 = diff * diff;
		let power = 1.0 / (n2.sqrt() * n2);
		f += diff*power;
	});
	*item_vi += f*DELTA_T;
});

position.iter().zip(velocity.iter_mut())
  .for_each(|(item_pi, item_vi)| {
	let mut f: Vector<Precision> = Vector::new(0.0, 0.0, 0.0);

	position.iter().for_each(|item_pj| {
		// Newton’s law of universal gravity calculation.
		let diff = *item_pj - *item_pi;
		let n2 = diff * diff;
		let power = 1.0 / (n2.sqrt() * n2);
		f += diff*power;
	});

	*item_vi += f*DELTA_T;
});

position.par_iter().zip(velocity.par_iter_mut())
  .for_each(|(item_pi, item_vi)| {
	let mut f: Vector<Precision> = Vector::new(0.0, 0.0, 0.0);

	position.iter().for_each(|item_pj| {
		// Newton’s law of universal gravity calculation.
		let diff = *item_pj - *item_pi;
		let n2 = diff.square();
		let power = 1.0 / (n2.sqrt() * n2);
		f += diff*power;
	});
	*item_vi += f*DELTA_T;
});

#[derive(Clone, Debug, PartialEq)]
pub struct NBody {
	pub position: Vec<Vector<Precision>>,
	pub velocity: Vec<Vector<Precision>>
}

typedef float Scalar;

typedef struct {
	Scalar x[nParticles];
	Scalar y[nParticles];
	Scalar z[nParticles];
	Scalar vx[nParticles];
	Scalar vy[nParticles];
	Scalar vz[nParticles];
} NBody;

// Entspricht direkter Assembler-Progrommierung
__m256 _mm256_add_ps (__m256 a, __m256 b);

// Vektor-Addition mit 8 einfach, genaue Fließkommazahlen
pub fn sum(a: f32x8, b: f32x8) -> f32x8 {
	a+b
}

pub struct Array<T>([T; N_PARTICLES_SOA]);

pub struct StructOfArrays<T> {
	pub x: Array<T>,
	pub y: Array<T>,
	pub z: Array<T>
}

pub struct NBodySoA {
	position: StructOfArrays<PrecisionSoA>,
	velocity: StructOfArrays<PrecisionSoA>
}

pub struct StructOfArraysIter<'a, T: 'a> {
	inner: &'a StructOfArrays<T>,
	pos: usize, len: usize
}

impl<'a, T> Iterator for StructOfArraysIter<'a, T> {
	type Item = (&'a T, &'a T, &'a T);
	fn next(&mut self) -> Option<Self::Item> {
		if self.pos < self.len {
			let result = (&self.inner.x.0[self.pos],
				      &self.inner.y.0[self.pos],
				      &self.inner.z.0[self.pos]);
			self.pos += 1;
			Some(result)
	} } }

position.iter().zip(velocity.iter_mut())
  .for_each(|((pix, piy, piz), (vix, viy, viz))| {
	let mut fx: PrecisionSoA = ...
	position.iter().for_each(|(pjx, pjy, pjz)| {
		let mut dx: ...
		for lane in 0..PrecisionSoA::lanes() {
			dx +=  *pjx - PrecisionSoA::splat(pix.extract(lane));
			dy +=  ..
		}
		let n2 = dx*dx + dy*dy + dz*dz;
		let power = 1.0 / (n2.sqrt() * n2);
		fx += dx*power; fy += dy*power; fz += dz*power;
	});
	*vix += fx * dt; *viy += fy * dt; *viz += fz * dt;
});

position.iter().zip(velocity.iter_mut()).par_bridge()
  .for_each(|((pix, piy, piz), (vix, viy, viz))| {
	..
}

let listener = TcpListener::bind(&addr).unwrap();
let server = listener
    .incoming()
    .for_each(|socket| {
        // für jede Verbindung
        // siehe nächste Folie
    })
    .map_err(|err| {
        // Fehlerbehandlung
        println!("accept error = {:?}", err);
    });
tokio::run(server);

let (reader, writer) = socket.split();
let amountF = io::copy(reader, writer);

let msg = amountF.then(|result| {
    match result {
        Ok((amount, _, _)) => println!("wrote {} bytes", amount),
        Err(e) => println!("error: {}", e),
    }

    Ok(())
});

tokio::spawn(msg);
Ok(())

TcpStream::connect(&addr)
    .and_then(|stream| {
        println!("connected");

        io::write_all(stream, "hello para//el\n").then(|result| {
            println!("wrote to connection; success={:?}",
                     result.is_ok());
            Ok(())
        })
    })
    .map_err(|err| {
        println!("connection error = {:?}", err);
    })

let number_of_connections = 100_000;
let retry_strategy = ExponentialBackoff::from_millis(10)
                     .map(jitter).take(3);

let client = stream::iter_ok(0..number_of_connections)
    .for_each(move |_| Retry::spawn(
        retry_strategy.clone(),
        action
    ).then(|_| Ok(())));

fn main() {
    let closure = async || {
         println("Hello from async closure.");
    };
    println!("Hello from main");
    let future = closure();
    println!("Hello from main again");
    futures::block_on(future);
}