seq/seq.go

   1 // Copyright 2010 Bill Burdick. All rights reserved.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4 //
   5 package seq
   6
   7 import "fmt"
   8 import "io"
   9 import "os"
  10 import "container/vector"
  11
  12 type El interface{}
  13 type Seq interface {
  14         // core methods
  15         Find(f func(i El)bool) El
  16         Rest() Seq
  17         Len() int
  18         // wrappers that keep the current type
  19         Append(s2 Seq) Seq
  20         Prepend(s2 Seq) Seq
  21         Filter(filter func(e El) bool) Seq
  22         Map(f func(i El) El) Seq
  23         FlatMap(f func(i El) Seq) Seq
  24 }
  25
  26 //convert a sequence to a concurrent sequence
  27 func Concurrent(s Seq) ConcurrentSeq {
  28         switch seq := s.(type) {case ConcurrentSeq: return seq}
  29         return Gen(func(c SeqChan){Output(s, c)})
  30 }
  31
  32 //convert a sequence to a sequential sequence
  33 func Sequential(s Seq) *SequentialSeq {
  34         switch seq := s.(type) {case *SequentialSeq: return seq}
  35         return SMap(s, func(el El)El{return el})
  36 }
  37
  38 func FirstN(s Seq, n int) []interface{} {
  39         r := make([]interface{}, n)
  40         x := 0
  41         Find(s, func(el El)bool{
  42                 r[x] = el
  43                 x++
  44                 return x == n
  45         })
  46         return r
  47 }
  48
  49 func First2(s Seq) (a, b interface{}) {
  50         r := FirstN(s, 2)
  51         return r[0], r[1]
  52 }
  53
  54 func First3(s Seq) (a, b, c interface{}) {
  55         r := FirstN(s, 3)
  56         return r[0], r[1], r[2]
  57 }
  58
  59 func First4(s Seq) (a, b, c, d interface{}) {
  60         r := FirstN(s, 4)
  61         return r[0], r[1], r[2], r[3]
  62 }
  63
  64 func First5(s Seq) (a, b, c, d, e interface{}) {
  65         r := FirstN(s, 5)
  66         return r[0], r[1], r[2], r[3], r[4]
  67 }
  68
  69 func First6(s Seq) (a, b, c, d, e, f interface{}) {
  70         r := FirstN(s, 6)
  71         return r[0], r[1], r[2], r[3], r[4], r[5]
  72 }
  73
  74 func IsSeq(s interface{}) bool {
  75         _, test := s.(Seq)
  76         return test
  77 }
  78
  79 func First(s Seq) interface{} {
  80         var result interface{}
  81         s.Find(func(el El)bool{
  82                 result = el
  83                 return true
  84         })
  85         return result
  86 }
  87
  88 func IsEmpty(s Seq) bool {
  89         empty := true
  90         s.Find(func(el El)bool{
  91                 empty = false
  92                 return true
  93         })
  94         return empty
  95 }
  96
  97 func Find(s Seq, f func(el El) bool) El {return s.Find(f)}
  98
  99 func While(s Seq, f func(el El) bool) {s.Find(func(el El)bool{return !f(el)})}
 100
 101 func Do(s Seq, f func(el El)) {
 102         s.Find(func(el El)bool{
 103                 f(el)
 104                 return false
 105         })
 106 }
 107
 108 // CDo -- do f concurrently on each element of s, in any order
 109 func CDo(s Seq, f func(el El)) {
 110         c := CMap(s, func(el El)El{f(el); return nil})()
 111         for <- c; !closed(c); <- c {}
 112 }
 113
 114 func Len(s Seq) int {return s.Len()}
 115
 116 func Output(s Seq, c SeqChan) {
 117         Do(s, func(el El){c <- el})
 118 }
 119
 120 func Rest(s Seq) Seq {return s.Rest()}
 121
 122 func Append(s1 Seq, s2 Seq) Seq {return s1.Append(s2)}
 123
 124 func AppendToVector(s Seq, vec *vector.Vector) {
 125         switch arg := s.(type) {
 126         case *SequentialSeq: vec.AppendVector((*vector.Vector)(arg))
 127         default: Do(s, func(el El){vec.Push(el)})
 128         }
 129 }
 130
 131 func SAppend(s Seq, s2 Seq) *SequentialSeq {
 132         vec := make(vector.Vector, 0, quickLen(s, 8) + quickLen(s2, 8))
 133         AppendToVector(s, &vec)
 134         AppendToVector(s2, &vec)
 135         return (*SequentialSeq)(&vec)
 136 }
 137
 138 func CAppend(s Seq, s2 Seq) ConcurrentSeq {
 139         return Gen(func(c SeqChan){
 140                 Output(s, c)
 141                 Output(s2, c)
 142         })
 143 }
 144
 145 func Prepend(s1 Seq, s2 Seq) Seq {return s1.Prepend(s2)}
 146
 147 func quickLen(s Seq, d int) int {
 148         switch seq := s.(type) {case *SequentialSeq: return s.Len()}
 149         return d
 150 }
 151
 152 func Filter(s Seq, filter func(e El)bool) Seq {return s.Filter(filter)}
 153
 154 func ifFunc(condition func(e El)bool, op func(e El)) func(el El){return func(el El){if condition(el) {op(el)}}}
 155
 156 func SFilter(s Seq, filter func(e El)bool) *SequentialSeq {
 157         //continue shrinking
 158         vec := make(vector.Vector, 0, quickLen(s, 8))
 159         Do(s, ifFunc(filter, func(el El){vec.Push(el)}))
 160         return (*SequentialSeq)(&vec)
 161 }
 162
 163 func CFilter(s Seq, filter func(e El)bool) ConcurrentSeq {
 164         return Gen(func(c SeqChan){
 165                 Do(s, ifFunc(filter, func(el El){c <- el}))
 166         })
 167 }
 168
 169 func Map(s Seq, f func(el El) El) Seq {return s.Map(f)}
 170
 171 func SMap(s Seq, f func(i El)El) *SequentialSeq {
 172         vec := make(vector.Vector, 0, quickLen(s, 8))
 173         Do(s, func(el El){vec.Push(f(el))})
 174         return (*SequentialSeq)(&vec)
 175 }
 176
 177 type reply struct {
 178         index int;
 179         result El
 180 }
 181
 182 type swEntry struct {
 183         value El
 184         present bool
 185 }
 186
 187 // SlidingWindow is a vector with limited capacity (power of 2) and a base
 188 type SlidingWindow struct {
 189         start, base, count, mask int
 190         values []swEntry
 191 }
 192 // NewSlidingWindow creates a new SlidingWindow with capacity size
 193 func NewSlidingWindow(sz uint) *SlidingWindow {return &SlidingWindow{0, 0, 0, (1 << sz) - 1, make([]swEntry, 1 << sz)}}
 194 func (r *SlidingWindow) Max() int {return r.base + r.Capacity()}
 195 func (r *SlidingWindow) Capacity() int {return len(r.values)}
 196 func (r *SlidingWindow) normalize(index int) int {return (index + r.Capacity()) & r.mask}
 197 func (r *SlidingWindow) IsEmpty() bool {return r.count == 0}
 198 func (r *SlidingWindow) IsFull() bool {return r.count == r.Capacity()}
 199 func (r *SlidingWindow) GetFirst() (interface{}, bool) {return r.values[r.start].value, r.values[r.start].present}
 200 func (r *SlidingWindow) RemoveFirst() (interface{}, bool) {
 201         result := r.values[r.start]
 202         if !result.present {return nil, false}
 203         r.values[r.start] = swEntry{nil, false}
 204         r.count--
 205         r.start = r.normalize(r.start + 1)
 206         r.base++
 207         return result.value, true
 208 }
 209 func (r *SlidingWindow) Get(index int) (interface{}, bool) {
 210         index -= r.base
 211         if index < 0 || index >= r.Capacity() {return nil, false}
 212         index = r.normalize(index + r.start)
 213         value := r.values[index]
 214         return value.value, value.present
 215 }
 216 func (r *SlidingWindow) Set(index int, value interface{}) bool {
 217         index -= r.base
 218         if index < 0 || index >= r.Capacity() {return false}
 219         index = r.normalize(index + r.start)
 220         r.values[index].value = value
 221         if !r.values[index].present {
 222                 r.values[index].present = true
 223                 r.count++
 224         }
 225         return true
 226 }
 227
 228 // spawn a goroutine that does the following for each value, with up to size pending at a time:
 229 //   spawn a goroutine to apply f to the value and send the result back in a channel
 230 // send the results in order to the ouput channel as they are completed
 231 func CMap(s Seq, f func(el El) El, sizePowerOpt... uint) ConcurrentSeq {
 232         sizePower := uint(6)
 233         if len(sizePowerOpt) > 0 {sizePower = sizePowerOpt[0]}
 234         size := 1 << sizePower
 235         return Gen(func(output SeqChan){
 236                 //punt and convert sequence to concurrent
 237                 //maybe someday we'll handle SequentialSequences separately
 238                 input := Concurrent(s)()
 239                 window := NewSlidingWindow(sizePower)
 240                 replyChannel := make(chan reply)
 241                 inputCount, pendingInput, pendingOutput := 0, 0, 0
 242                 inputClosed := false
 243                 defer close(replyChannel)
 244                 for !inputClosed || pendingInput > 0 || pendingOutput > 0 {
 245                         first, hasFirst := window.GetFirst()
 246                         ic, oc, rc := input, output, replyChannel
 247                         if !hasFirst {oc = nil}
 248                         if inputClosed || pendingInput >= size {ic = nil}
 249                         if pendingOutput >= size {rc = nil}
 250                         select {
 251                         case oc <- first:
 252                                 window.RemoveFirst()
 253                                 pendingOutput--
 254                         case inputElement := <- ic:
 255                                 if closed(ic) {
 256                                         inputClosed = true
 257                                 } else {
 258                                         go func(index int, value interface{}) {
 259                                                 replyChannel <- reply{index, f(value)}
 260                                         }(inputCount, inputElement)
 261                                         inputCount++
 262                                         pendingInput++
 263                                 }
 264                         case replyElement := <- rc:
 265                                 window.Set(replyElement.index, replyElement.result)
 266                                 pendingInput--
 267                                 pendingOutput++
 268                         }
 269                 }
 270         })
 271 }
 272
 273 func FlatMap(s Seq, f func(el El) Seq) Seq {return s.FlatMap(f)}
 274
 275 func SFlatMap(s Seq, f func(i El) Seq) *SequentialSeq {
 276         vec := make(vector.Vector, 0, quickLen(s, 8))
 277         Do(s, func(e El){Do(f(e).(Seq), func(sub El){vec.Push(sub)})})
 278         return (*SequentialSeq)(&vec)
 279 }
 280
 281 func CFlatMap(s Seq, f func(i El) Seq, sizeOpt... uint) ConcurrentSeq {
 282         return Gen(func(c SeqChan){
 283                 Do(CMap(s, func(e El)El{return f(e)}, sizeOpt...), func(sub El){
 284                         Output(sub.(Seq), c)
 285                 })
 286         })
 287 }
 288
 289 func Fold(s Seq, init interface{}, f func(acc, el El)El) interface{} {
 290         Do(s, func(el El){init = f(init, el)})
 291         return init
 292 }
 293
 294 //maybe convert this to use an accumulator instead of append?
 295 func Combinations(s Seq, number int) Seq {
 296         if number == 0 || IsEmpty(s) {return From(From())}
 297         return Combinations(s.Rest(), number).Prepend(Combinations(s.Rest(), number - 1).Map(func(el El)El{
 298                 return el.(Seq).Prepend(From(First(s)))
 299         }))
 300 }
 301
 302 //var Names map[interface{}]string
 303
 304 //returns the product of the Seqs contained in sequences
 305 func Product(sequences Seq) Seq {
 306         return Fold(sequences, From(From()), func(result, each El)El{
 307                 return result.(Seq).FlatMap(func(seq El)Seq{
 308                         return each.(Seq).Map(func(i El) El {
 309                                 return seq.(Seq).Append(From(i))
 310                         })
 311                 })
 312         }).(Seq)
 313 }
 314
 315 func Prettyln(s interface{}, rest... interface{}) {
 316         writer := Pretty(s, rest...)
 317         fmt.Fprintln(writer)
 318 }
 319 func Pretty(s interface{}, args... interface{}) io.Writer {
 320         var writer io.Writer = os.Stdout
 321         var names map[interface{}]string
 322         for i := 0; i < len(args); i++ {
 323                 switch arg := args[i].(type) {
 324                 case map[interface{}]string: names = arg
 325                 case io.Writer: writer = arg
 326                 }
 327         }
 328         if names == nil {names = map[interface{}]string{}}
 329         prettyLevel(s, 0, names, writer)
 330         return writer
 331 }
 332
 333 //This pretty is ugly :)
 334 func prettyLevel(s interface{}, level int, names map[interface{}]string, w io.Writer) {
 335         name, hasName := names[s]
 336         if hasName {
 337                 fmt.Fprint(w, name)
 338         } else switch arg := s.(type) {
 339         case Seq:
 340                 fmt.Fprintf(w, "%*s%s", level, "", "[")
 341                 first := true
 342                 innerSeq := false
 343                 named := false
 344                 Do(arg, func(v El) {
 345                         _, named = names[v]
 346                         _,innerSeq = v.(Seq)
 347                         if first {
 348                                 first = false
 349                                 if !named && innerSeq {
 350                                         fmt.Fprintln(w)
 351                                 }
 352                         } else if !named && innerSeq {
 353                                 fmt.Fprintln(w, ",")
 354                         } else {
 355                                 fmt.Fprint(w, ", ")
 356                         }
 357                         if innerSeq {
 358                                 prettyLevel(v.(Seq), level + 4, names, w)
 359                         } else {
 360                                 fmt.Fprintf(w, "%v", v)
 361                         }
 362                 })
 363                 if innerSeq {
 364                         if !named {
 365                                 fmt.Fprintf(w, "\n%*s", level, "")
 366                         }
 367                 }
 368                 fmt.Fprintf(w, "]")
 369         default:
 370                 fmt.Print(arg)
 371         }
 372 }
 373
 374 //ConcurrentSeq
 375
 376 type SeqChan chan interface{}
 377
 378 type ConcurrentSeq func()SeqChan
 379
 380 // f must behave properly when the channel is closed, so that IsEmpty and First work properly
 381 func Gen(f func(c SeqChan)) ConcurrentSeq {
 382         return func() SeqChan {
 383                 c := make(SeqChan)
 384                 go func() {
 385                         defer close(c)
 386                         f(c)
 387                 }()
 388                 return c
 389         }
 390 }
 391
 392 func CUpto(limit int) ConcurrentSeq {
 393         return Gen(func(c SeqChan) {
 394                 for i := 0; i < limit; i++ {
 395                         c <- i
 396                 }
 397         })
 398 }
 399
 400 func (s ConcurrentSeq) Find(f func(el El)bool) El {
 401         c := s()
 402         defer close(c)
 403         for el := <- c; !closed(c) ; el = <- c {
 404                 if f(el) {return el}
 405         }
 406         return nil
 407 }
 408
 409 func (s ConcurrentSeq) Rest() Seq {
 410         return ConcurrentSeq(func()SeqChan{
 411                 c := s()
 412                 <- c
 413                 return c
 414         })
 415 }
 416
 417 func (s ConcurrentSeq) Len() int {
 418         len := 0
 419         Do(s, func(el El){
 420                 len++
 421         })
 422         return len
 423 }
 424
 425 func (s ConcurrentSeq) Append(s2 Seq) Seq {return CAppend(s, s2)}
 426
 427 func (s ConcurrentSeq) Prepend(s2 Seq) Seq {return CAppend(s2, s)}
 428
 429 func (s ConcurrentSeq) Filter(f func(e El)bool) Seq {return CFilter(s, f)}
 430
 431 func (s ConcurrentSeq) Map(f func(i El)El) Seq {return CMap(s, f)}
 432
 433 func (s ConcurrentSeq) FlatMap(f func(i El) Seq) Seq {return CFlatMap(s, f)}
 434
 435 // recursively convert nested concurrent sequences to sequential
 436 // does not descend into nested sequential sequences
 437 func (s ConcurrentSeq) ToSequentialSeq() *SequentialSeq {
 438         return SMap(s, func(el El)El{
 439                 switch seq := el.(type) {case ConcurrentSeq: return seq.ToSequentialSeq()}
 440                 return el
 441         })
 442 }
 443
 444
 445 // SequentialSeq
 446
 447 type SequentialSeq []interface{}
 448
 449 func From(els... interface{}) *SequentialSeq {return (*SequentialSeq)(&els)}
 450
 451 func AUpto(limit int) *SequentialSeq {
 452         a := make([]interface{}, limit)
 453         for i := 0; i < limit; i++ {
 454                 a[i] = i
 455         }
 456         return (*SequentialSeq)(&a)
 457 }
 458
 459 func (s *SequentialSeq) Find(f func(el El)bool) El {
 460         for i := 0; i < len(*s); i++ {
 461                 if f((*s)[i]) {return (*s)[i]}
 462         }
 463         return nil
 464 }
 465
 466 func (s *SequentialSeq) Rest() Seq {
 467         s2 := (*s)[1:]
 468         return (*SequentialSeq)(&s2)
 469 }
 470
 471 func (s *SequentialSeq) Len() int {return len(*s)}
 472
 473 func (s *SequentialSeq) Append(s2 Seq) Seq {return SAppend(s, s2)}
 474
 475 func (s *SequentialSeq) Prepend(s2 Seq) Seq {return SAppend(s2, s)}
 476
 477 func (s *SequentialSeq) Filter(f func(e El)bool) Seq {return SFilter(s, f)}
 478
 479 func (s *SequentialSeq) Map(f func(i El)El) Seq {return SMap(s, f)}
 480
 481 func (s *SequentialSeq) FlatMap(f func(i El) Seq) Seq {return SFlatMap(s, f)}