Skip to content

Instantly share code, notes, and snippets.

@stef
Created March 8, 2021 16:24
Show Gist options
  • Save stef/5e2d8a88f1d9d220f623a55531a9fb06 to your computer and use it in GitHub Desktop.
Save stef/5e2d8a88f1d9d220f623a55531a9fb06 to your computer and use it in GitHub Desktop.
gnuradio3.8/gr-clenabled/intel-opencl-icd runtime crash test-cases
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: GPL-3.0
#
# GNU Radio Python Flow Graph
# Title: trigger cl racecondition
# GNU Radio version: 3.8.1.0
from gnuradio import analog
from gnuradio import blocks
from gnuradio import gr
from gnuradio.filter import firdes
import sys
import signal
from argparse import ArgumentParser
from gnuradio.eng_arg import eng_float, intx
from gnuradio import eng_notation
import clenabled
class racecond(gr.top_block):
def __init__(self, samp_rate=1e6):
gr.top_block.__init__(self, "trigger cl racecondition")
##################################################
# Parameters
##################################################
self.samp_rate = samp_rate
##################################################
# Blocks
##################################################
self.clenabled_clMultiply_0_1 = clenabled.clMathOp(1,1,1,0,0,1,0)
self.clenabled_clMultiply_0 = clenabled.clMathOp(1,1,1,0,0,1,0)
self.blocks_throttle_0_0 = blocks.throttle(gr.sizeof_gr_complex*1, samp_rate,True)
self.blocks_throttle_0 = blocks.throttle(gr.sizeof_gr_complex*1, samp_rate,True)
self.blocks_null_sink_0 = blocks.null_sink(gr.sizeof_gr_complex*1)
self.analog_sig_source_x_0_0 = analog.sig_source_c(samp_rate, analog.GR_SIN_WAVE, 0, 1, 0, 0)
self.analog_sig_source_x_0 = analog.sig_source_c(samp_rate, analog.GR_COS_WAVE, 0, 1, 0, 0)
##################################################
# Connections
##################################################
self.connect((self.analog_sig_source_x_0, 0), (self.blocks_throttle_0, 0))
self.connect((self.analog_sig_source_x_0_0, 0), (self.blocks_throttle_0_0, 0))
self.connect((self.blocks_throttle_0, 0), (self.clenabled_clMultiply_0, 1))
self.connect((self.blocks_throttle_0, 0), (self.clenabled_clMultiply_0_1, 1))
self.connect((self.blocks_throttle_0_0, 0), (self.clenabled_clMultiply_0, 0))
self.connect((self.blocks_throttle_0_0, 0), (self.clenabled_clMultiply_0_1, 0))
self.connect((self.clenabled_clMultiply_0, 0), (self.blocks_null_sink_0, 0))
self.connect((self.clenabled_clMultiply_0_1, 0), (self.blocks_null_sink_0, 1))
def get_samp_rate(self):
return self.samp_rate
def set_samp_rate(self, samp_rate):
self.samp_rate = samp_rate
self.analog_sig_source_x_0.set_sampling_freq(self.samp_rate)
self.analog_sig_source_x_0_0.set_sampling_freq(self.samp_rate)
self.blocks_throttle_0.set_sample_rate(self.samp_rate)
self.blocks_throttle_0_0.set_sample_rate(self.samp_rate)
def argument_parser():
parser = ArgumentParser()
return parser
def main(top_block_cls=racecond, options=None):
if options is None:
options = argument_parser().parse_args()
tb = top_block_cls()
def sig_handler(sig=None, frame=None):
tb.stop()
tb.wait()
sys.exit(0)
signal.signal(signal.SIGINT, sig_handler)
signal.signal(signal.SIGTERM, sig_handler)
tb.start()
try:
input('Press Enter to quit: ')
except EOFError:
pass
tb.stop()
tb.wait()
if __name__ == '__main__':
main()
options:
parameters:
author: ''
category: '[GRC Hier Blocks]'
cmake_opt: ''
comment: ''
copyright: ''
description: ''
gen_cmake: 'On'
gen_linking: dynamic
generate_options: no_gui
hier_block_src_path: '.:'
id: racecond
max_nouts: '0'
output_language: python
placement: (0,0)
qt_qss_theme: ''
realtime_scheduling: ''
run: 'True'
run_command: '{python} -u {filename}'
run_options: prompt
sizing_mode: fixed
thread_safe_setters: ''
title: trigger cl racecondition
window_size: ''
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [8, 8]
rotation: 0
state: enabled
blocks:
- name: analog_sig_source_x_0
id: analog_sig_source_x
parameters:
affinity: ''
alias: ''
amp: '1'
comment: ''
freq: '0'
maxoutbuf: '0'
minoutbuf: '0'
offset: '0'
phase: '0'
samp_rate: samp_rate
type: complex
waveform: analog.GR_COS_WAVE
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [96, 172.0]
rotation: 0
state: true
- name: analog_sig_source_x_0_0
id: analog_sig_source_x
parameters:
affinity: ''
alias: ''
amp: '1'
comment: ''
freq: '0'
maxoutbuf: '0'
minoutbuf: '0'
offset: '0'
phase: '0'
samp_rate: samp_rate
type: complex
waveform: analog.GR_SIN_WAVE
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [120, 476.0]
rotation: 0
state: true
- name: blocks_null_sink_0
id: blocks_null_sink
parameters:
affinity: ''
alias: ''
bus_structure_sink: '[[0,],]'
comment: ''
num_inputs: '2'
type: complex
vlen: '1'
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [1312, 356.0]
rotation: 0
state: true
- name: blocks_throttle_0
id: blocks_throttle
parameters:
affinity: ''
alias: ''
comment: ''
ignoretag: 'True'
maxoutbuf: '0'
minoutbuf: '0'
samples_per_second: samp_rate
type: complex
vlen: '1'
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [416, 228.0]
rotation: 0
state: true
- name: blocks_throttle_0_0
id: blocks_throttle
parameters:
affinity: ''
alias: ''
comment: ''
ignoretag: 'True'
maxoutbuf: '0'
minoutbuf: '0'
samples_per_second: samp_rate
type: complex
vlen: '1'
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [464, 532.0]
rotation: 0
state: true
- name: clenabled_clMultiply_0
id: clenabled_clMultiply
parameters:
affinity: ''
alias: ''
comment: ''
deviceId: '0'
devices: '1'
maxoutbuf: '0'
minoutbuf: '0'
openCLPlatform: '1'
platformId: '0'
setDebug: '0'
type: complex
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [880, 208.0]
rotation: 0
state: enabled
- name: clenabled_clMultiply_0_1
id: clenabled_clMultiply
parameters:
affinity: ''
alias: ''
comment: ''
deviceId: '0'
devices: '1'
maxoutbuf: '0'
minoutbuf: '0'
openCLPlatform: '1'
platformId: '0'
setDebug: '0'
type: complex
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [880, 440.0]
rotation: 0
state: enabled
- name: samp_rate
id: parameter
parameters:
alias: ''
comment: ''
hide: none
label: ''
short_id: ''
type: ''
value: 1e6
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [280, 24.0]
rotation: 0
state: true
connections:
- [analog_sig_source_x_0, '0', blocks_throttle_0, '0']
- [analog_sig_source_x_0_0, '0', blocks_throttle_0_0, '0']
- [blocks_throttle_0, '0', clenabled_clMultiply_0, '1']
- [blocks_throttle_0, '0', clenabled_clMultiply_0_1, '1']
- [blocks_throttle_0_0, '0', clenabled_clMultiply_0, '0']
- [blocks_throttle_0_0, '0', clenabled_clMultiply_0_1, '0']
- [clenabled_clMultiply_0, '0', blocks_null_sink_0, '0']
- [clenabled_clMultiply_0_1, '0', blocks_null_sink_0, '1']
metadata:
file_format: 1
options:
parameters:
author: ''
category: '[GRC Hier Blocks]'
cmake_opt: ''
comment: ''
copyright: ''
description: ''
gen_cmake: 'On'
gen_linking: dynamic
generate_options: no_gui
hier_block_src_path: '.:'
id: racelpf
max_nouts: '0'
output_language: python
placement: (0,0)
qt_qss_theme: ''
realtime_scheduling: ''
run: 'True'
run_command: '{python} -u {filename}'
run_options: prompt
sizing_mode: fixed
thread_safe_setters: ''
title: testing clenabled racecondition with lpf
window_size: ''
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [8, 8]
rotation: 0
state: enabled
blocks:
- name: samp_rate
id: variable
parameters:
comment: ''
value: 1e6
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [280, 8.0]
rotation: 0
state: enabled
- name: analog_noise_source_x_0
id: analog_noise_source_x
parameters:
affinity: ''
alias: ''
amp: '1'
comment: ''
maxoutbuf: '0'
minoutbuf: '0'
noise_type: analog.GR_GAUSSIAN
seed: '0'
type: complex
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [144, 304.0]
rotation: 0
state: true
- name: blocks_null_sink_0
id: blocks_null_sink
parameters:
affinity: ''
alias: ''
bus_structure_sink: '[[0,],]'
comment: ''
num_inputs: '2'
type: complex
vlen: '1'
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [1840, 464.0]
rotation: 0
state: true
- name: blocks_throttle_0
id: blocks_throttle
parameters:
affinity: ''
alias: ''
comment: ''
ignoretag: 'True'
maxoutbuf: '0'
minoutbuf: '0'
samples_per_second: samp_rate
type: complex
vlen: '1'
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [408, 324.0]
rotation: 0
state: true
- name: clenabled_clLowPassFilter_0
id: clenabled_clLowPassFilter
parameters:
affinity: ''
alias: ''
beta: '6.76'
comment: ''
cutoff_freq: 75e3
decimation: '1'
deviceId: '0'
devices: '1'
gain: '1'
maxoutbuf: '0'
minoutbuf: '0'
openCLPlatform: '1'
platformId: '0'
samp_rate: samp_rate
setDebug: '0'
use_time: 'True'
width: 10e3
win: firdes.WIN_HAMMING
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [896, 104.0]
rotation: 0
state: true
- name: clenabled_clLowPassFilter_0_0
id: clenabled_clLowPassFilter
parameters:
affinity: ''
alias: ''
beta: '6.76'
comment: ''
cutoff_freq: 75e3
decimation: '1'
deviceId: '0'
devices: '1'
gain: '1'
maxoutbuf: '0'
minoutbuf: '0'
openCLPlatform: '1'
platformId: '0'
samp_rate: samp_rate
setDebug: '0'
use_time: 'True'
width: 10e3
win: firdes.WIN_HAMMING
states:
bus_sink: false
bus_source: false
bus_structure: null
coordinate: [904, 392.0]
rotation: 0
state: enabled
connections:
- [analog_noise_source_x_0, '0', blocks_throttle_0, '0']
- [blocks_throttle_0, '0', clenabled_clLowPassFilter_0, '0']
- [blocks_throttle_0, '0', clenabled_clLowPassFilter_0_0, '0']
- [clenabled_clLowPassFilter_0, '0', blocks_null_sink_0, '0']
- [clenabled_clLowPassFilter_0_0, '0', blocks_null_sink_0, '1']
metadata:
file_format: 1
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: GPL-3.0
#
# GNU Radio Python Flow Graph
# Title: testing clenabled racecondition with lpf
# GNU Radio version: 3.8.1.0
from gnuradio import analog
from gnuradio import blocks
from gnuradio import gr
from gnuradio.filter import firdes
import sys
import signal
from argparse import ArgumentParser
from gnuradio.eng_arg import eng_float, intx
from gnuradio import eng_notation
import clenabled
class racelpf(gr.top_block):
def __init__(self):
gr.top_block.__init__(self, "testing clenabled racecondition with lpf")
##################################################
# Variables
##################################################
self.samp_rate = samp_rate = 1e6
##################################################
# Blocks
##################################################
self.clenabled_clLowPassFilter_0_0 = clenabled.clFilter(1,1,0,0,1,firdes.low_pass(
1, samp_rate, 75e3, 10e3, firdes.WIN_HAMMING, 6.76),1,0,True)
self.clenabled_clLowPassFilter_0 = clenabled.clFilter(1,1,0,0,1,firdes.low_pass(
1, samp_rate, 75e3, 10e3, firdes.WIN_HAMMING, 6.76),1,0,True)
self.blocks_throttle_0 = blocks.throttle(gr.sizeof_gr_complex*1, samp_rate,True)
self.blocks_null_sink_0 = blocks.null_sink(gr.sizeof_gr_complex*1)
self.analog_noise_source_x_0 = analog.noise_source_c(analog.GR_GAUSSIAN, 1, 0)
##################################################
# Connections
##################################################
self.connect((self.analog_noise_source_x_0, 0), (self.blocks_throttle_0, 0))
self.connect((self.blocks_throttle_0, 0), (self.clenabled_clLowPassFilter_0, 0))
self.connect((self.blocks_throttle_0, 0), (self.clenabled_clLowPassFilter_0_0, 0))
self.connect((self.clenabled_clLowPassFilter_0, 0), (self.blocks_null_sink_0, 0))
self.connect((self.clenabled_clLowPassFilter_0_0, 0), (self.blocks_null_sink_0, 1))
def get_samp_rate(self):
return self.samp_rate
def set_samp_rate(self, samp_rate):
self.samp_rate = samp_rate
self.blocks_throttle_0.set_sample_rate(self.samp_rate)
self.clenabled_clLowPassFilter_0.set_taps2(firdes.low_pass(1, self.samp_rate, 75e3, 10e3, firdes.WIN_HAMMING, 6.76))
self.clenabled_clLowPassFilter_0_0.set_taps2(firdes.low_pass(1, self.samp_rate, 75e3, 10e3, firdes.WIN_HAMMING, 6.76))
def main(top_block_cls=racelpf, options=None):
tb = top_block_cls()
def sig_handler(sig=None, frame=None):
tb.stop()
tb.wait()
sys.exit(0)
signal.signal(signal.SIGINT, sig_handler)
signal.signal(signal.SIGTERM, sig_handler)
tb.start()
try:
input('Press Enter to quit: ')
except EOFError:
pass
tb.stop()
tb.wait()
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment