signal_processing/moving_average_parallel.jl_back at main · MozammilQ/signal_processing · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

using Distributed
addprocs(3)

@everywhere procs() using SharedArrays
M=3
function mov_avg_serial(arr::Array{Float64,1},M::Int)
	@info "Local serial Array{Float64,1} invoked"
	new_arr=Array{Float64,1}(undef,length(arr)-M+1)
	for i in 1:1:length(new_arr)
		for j in i:1:i+M-1
			new_arr[i]+=arr[j]
		end
	end
	return new_arr
end

function mov_avg_serial(arr::SharedArray{Float64,1},M::Int)
	@info "Local serial SharedArray{Float64,1} invoked"
	new_arr=Array{Float64,1}(undef,length(arr)-M+1)
	for i in 1:1:length(new_arr)
		for j in i:1:i+M-1
			new_arr[i]+=arr[j]
		end
	end
	return new_arr
end

@everywhere workers() begin

	M=$M

	function mov_avg_parallel(M::Int,start_::Int,end_::Int)
		@info "SharedArray{Float64,1} invoked at worker " * string(myid())
	        new_arr=Array{Float64,1}(undef,end_-start_-M+1)
	        for i in 1:1:end_-start_-M+1
	                for j in i:1:i+M-1
	                        new_arr[i]+=arr[start_+j-1]
	                end
	        end
	        return new_arr
	end

	function mov_avg_parallel(M::Int,start_::Int,end_::Int)
		@info "Array{Float64,1} invoked at worker " * string(myid())
	        new_arr=Array{Float64,1}(undef,end_-start_-M+1)
	        for i in 1:1:end_-start_-M+1
	                for j in i:1:i+M-1
	                        new_arr[i]+=arr[start_+j-1]
	                end
	        end
	        return new_arr
	end

end

function mov_avg_paralle(arr::Array{Float64},M::Int)
	@info "Local mov_avg_paralle Array invoked"
	marker=round(Int,length(arr)/nworkers())

	wrkrs=workers()
	result_=Float64[]

	for i in wrkrs

		start_= round(Int, marker * ( i - wrkrs[1] )) + 1
		end_=start_+marker+M-2


		if i==wrkrs[end]
			append!(result_,@fetchfrom i mov_avg_parallel(arr,M,start_,length(arr)))
		else
			append!(result_,@fetchfrom i mov_avg_parallel(arr,M,start_,end_))
		end

	end
	return result_
end

function mov_avg_paralle(arr::SharedArray{Float64},M::Int)
	@info "Local mov_avg_parralle SharedArray invoked"
	marker=round(Int,length(arr)/nworkers())

	wrkrs=workers()
	result_=Float64[]

	for i in wrkrs

		start_= round(Int, marker * ( i - wrkrs[1] )) + 1
		end_=start_+marker+M-2


		if i==wrkrs[end]
			append!(result_,@fetchfrom i mov_avg_parallel(M,start_,length(arr)))
		else
			append!(result_,@fetchfrom i mov_avg_parallel(M,start_,end_))
		end

	end
	return result_
end


##########################################################################################################

ele=5_000_000;
using SharedArrays
arr=SharedArray{Float64,1}(ele);
#arr=rand(1:90000000.0,ele);

using BenchmarkTools


@btime mov_avg_serial(arr,M);
@btime mov_avg_paralle(arr,M);